From 2ae2571b2605bfc7d62859861846561247a4034e Mon Sep 17 00:00:00 2001 From: mattcasters Date: Tue, 29 Sep 2026 21:57:56 +0200 Subject: [PATCH 1/2] Issue #2406 : Flush text file output every few seconds by default --- .../main/java/org/apache/hop/core/Const.java | 8 +- .../pipeline/transforms/textfileoutput.adoc | 2 + .../modules/ROOT/pages/variables.adoc | 8 +- .../compress/CompressionOutputStream.java | 7 + .../compress/CompressionOutputStreamTest.java | 21 ++ .../textfileoutput/TextFileOutput.java | 60 ++++-- .../textfileoutput/TextFileOutputData.java | 7 +- .../TextFileOutputFlushTest.java | 193 ++++++++++++++++++ 8 files changed, 287 insertions(+), 19 deletions(-) create mode 100644 plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java diff --git a/core/src/main/java/org/apache/hop/core/Const.java b/core/src/main/java/org/apache/hop/core/Const.java index e99cc8509ca..5c07ffa830e 100644 --- a/core/src/main/java/org/apache/hop/core/Const.java +++ b/core/src/main/java/org/apache/hop/core/Const.java @@ -811,13 +811,13 @@ public static boolean toBoolean(String string) { public static final String HOP_FILE_OUTPUT_MAX_STREAM_COUNT = "HOP_FILE_OUTPUT_MAX_STREAM_COUNT"; /** - * This variable contains the number of milliseconds between flushes of all open files in the Text - * File Output transform. + * Milliseconds between flushes of all open files in the Text File Output transform. {@code 0} + * selects the transform default of 5000. */ @Variable( - value = "0", + value = "5000", description = - "This project variable is used by the Text File Output transform. It defines the max number of milliseconds between flushes of files opened by the transform.") + "This project variable is used by the Text File Output transform. It defines how many milliseconds to wait between flushes of files opened by the transform. Output is buffered, so slow input stays invisible until the buffer fills or the file is closed. The default is 5000 (5 seconds). A value of 0 uses that default. Set a positive number of milliseconds to change the interval.") public static final String HOP_FILE_OUTPUT_MAX_STREAM_LIFE = "HOP_FILE_OUTPUT_MAX_STREAM_LIFE"; /** Set this variable to Y to disable standard Hop logging to the console. (stdout) */ diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc index 6a14d2e6d58..69e938aa04f 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc @@ -29,6 +29,8 @@ This is commonly used to generate Comma Separated Values (CSV files) that can be It is also possible to generate fixed width files by setting lengths on the fields in the fields tab. +Output is buffered. Open files are flushed on an interval so slow-arriving rows show up without waiting for that buffer to fill. The interval is xref:variables.adoc[HOP_FILE_OUTPUT_MAX_STREAM_LIFE], in milliseconds. The default is 5 seconds. A value of 0 uses that default. Set a positive number of milliseconds to change the interval. + You can choose to use a xref:metadata-types/static-schema-definition.adoc[Schema Definition] or to define the required fields' layout manually. If you decide to define the fields layout by using a xref:metadata-types/static-schema-definition.adoc[Schema Definition], use the xref:pipeline/transforms/schemamapping.adoc[Schema mapping] transform to adjust the incoming stream according to the chosen xref:metadata-types/static-schema-definition.adoc[Schema Definition] The `ignore manual fields` ignores any fields manually defined in the transform's field layout, and only uses the layout specified in the xref:metadata-types/static-schema-definition.adoc[Schema Definition]. diff --git a/docs/hop-user-manual/modules/ROOT/pages/variables.adoc b/docs/hop-user-manual/modules/ROOT/pages/variables.adoc index e8ac3c38ef2..37ec6165181 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/variables.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/variables.adoc @@ -258,8 +258,12 @@ Otherwise they are not. |HOP_FILE_OUTPUT_MAX_STREAM_COUNT|1024|This project variable is used by the Text File Output transform. It defines the max number of simultaneously open files within the transform. The transform will close/reopen files as necessary to insure the max is not exceeded -|HOP_FILE_OUTPUT_MAX_STREAM_LIFE|0|This project variable is used by the Text File Output transform. -It defines the max number of milliseconds between flushes of files opened by the transform. +|HOP_FILE_OUTPUT_MAX_STREAM_LIFE|5000|This project variable is used by the Text File Output transform. +It defines how many milliseconds to wait between flushes of files opened by the transform. +Output is buffered, so slow input stays invisible until the buffer fills or the file is closed. +The default is 5000 (5 seconds). +A value of 0 uses that default. +Set a positive number of milliseconds to change the interval. |HOP_GLOBAL_LOG_VARIABLES_CLEAR_ON_EXPORT|N|Set this variable to N to preserve global log variables defined in pipeline / workflow Properties -> Log panel. Changing it to true will clear it when export pipeline / workflow. |HOP_JSON_INPUT_INCLUDE_NULLS|Y|Name of te variable to set so that Nulls are considered while parsing JSON files. If HOP_JSON_INPUT_INCLUDE_NULLS is "Y" then nulls will be included otherwise they will not be included (default behavior) diff --git a/engine/src/main/java/org/apache/hop/core/compress/CompressionOutputStream.java b/engine/src/main/java/org/apache/hop/core/compress/CompressionOutputStream.java index 4c15cd9470a..3753c26327a 100644 --- a/engine/src/main/java/org/apache/hop/core/compress/CompressionOutputStream.java +++ b/engine/src/main/java/org/apache/hop/core/compress/CompressionOutputStream.java @@ -48,6 +48,13 @@ public void close() throws IOException { delegate.close(); } + @Override + public void flush() throws IOException { + if (delegate != null) { + delegate.flush(); + } + } + @Override public void write(int b) throws IOException { delegate.write(b); diff --git a/engine/src/test/java/org/apache/hop/core/compress/CompressionOutputStreamTest.java b/engine/src/test/java/org/apache/hop/core/compress/CompressionOutputStreamTest.java index def2e2e9ec7..939d59c2cb2 100644 --- a/engine/src/test/java/org/apache/hop/core/compress/CompressionOutputStreamTest.java +++ b/engine/src/test/java/org/apache/hop/core/compress/CompressionOutputStreamTest.java @@ -80,6 +80,18 @@ void testWrite() throws IOException { outStream.write("Test".getBytes()); } + @Test + void testFlushDelegates() throws IOException { + ICompressionProvider provider = outStream.getCompressionProvider(); + FlushCountStream out = new FlushCountStream(); + outStream = new DummyCompressionOS(out, provider); + outStream.write("Test".getBytes()); + assertEquals(0, out.flushes); + outStream.flush(); + assertEquals(1, out.flushes); + assertEquals("Test", out.toString()); + } + @Test void testAddEntry() throws IOException { ICompressionProvider provider = outStream.getCompressionProvider(); @@ -94,4 +106,13 @@ private static class DummyCompressionOS extends CompressionOutputStream { super(out, provider); } } + + private static class FlushCountStream extends ByteArrayOutputStream { + private int flushes; + + @Override + public void flush() { + flushes++; + } + } } diff --git a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java index 80df8b58106..78e4ca2e1d9 100644 --- a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java +++ b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java @@ -22,7 +22,6 @@ import java.io.OutputStream; import java.io.UnsupportedEncodingException; import java.util.ArrayList; -import java.util.Date; import java.util.List; import org.apache.commons.vfs2.FileObject; import org.apache.hop.core.Const; @@ -289,15 +288,19 @@ public String getOutputFileName(Object[] row) throws HopException { return filename; } + /** + * Milliseconds between flushes when {@link Const#HOP_FILE_OUTPUT_MAX_STREAM_LIFE} is unset or not + * a positive number. A few seconds, so slow input shows up without waiting for the buffer to + * fill. + */ + static final int DEFAULT_FILE_FLUSH_INTERVAL_MS = 5000; + public int getFlushInterval() { - String maxStreamLife = variables.getVariable("HOP_FILE_OUTPUT_MAX_STREAM_LIFE"); - int flushInterval = 0; - if (maxStreamLife != null) { - try { - flushInterval = Integer.parseInt(maxStreamLife); - } catch (Exception ex) { - // Do nothing - } + String maxStreamLife = variables.getVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE); + int flushInterval = Const.toInt(maxStreamLife, DEFAULT_FILE_FLUSH_INTERVAL_MS); + // 0 is the historical default and means "not configured". + if (flushInterval <= 0) { + return DEFAULT_FILE_FLUSH_INTERVAL_MS; } return flushInterval; } @@ -388,16 +391,16 @@ private boolean writeRowToFile(Object[] row) throws HopException { int flushInterval = getFlushInterval(); if (flushInterval > 0) { - long currentTime = new Date().getTime(); + long currentTime = currentFlushTimeMillis(); if (data.lastFileFlushTime == 0) { data.lastFileFlushTime = currentTime; - } else if (data.lastFileFlushTime - currentTime > flushInterval) { + } else if (currentTime - data.lastFileFlushTime > flushInterval) { try { data.getFileStreamsCollection().flushOpenFiles(false); } catch (IOException e) { throw new HopException("Unable to flush open files", e); } - data.lastFileFlushTime = new Date().getTime(); + data.lastFileFlushTime = currentTime; } } return true; @@ -452,6 +455,11 @@ private boolean writeRowToFile(Object[] row) throws HopException { } } + /** Clock for the file-flush interval. Tests advance this instead of sleeping. */ + protected long currentFlushTimeMillis() { + return System.currentTimeMillis(); + } + public void flushOpenFiles(boolean closeAfterFlush) throws IOException { TextFileOutputData.IFileStreamsCollection coll = data.getFileStreamsCollection(); if (coll != null) { @@ -525,6 +533,7 @@ public void writeRow(IRowMeta rowMeta, Object[] r) throws HopTransformException } incrementLinesOutput(); + markCurrentFileDirty(); } catch (Exception e) { throw new HopTransformException("Error writing line", e); @@ -732,6 +741,7 @@ protected boolean writeEndedLine() { if (sLine != null && !sLine.trim().isEmpty()) { data.writer.write(getBinaryString(sLine)); incrementLinesOutput(); + markCurrentFileDirty(); } } catch (Exception e) { logError("Error writing ended tag line: " + e.toString()); @@ -810,9 +820,35 @@ && containsSeparatorOrEnclosure( retval = true; } incrementLinesOutput(); + markCurrentFileDirty(); return retval; } + /** + * An interval flush clears the dirty flag. Later rows still land in the buffer, so the flag has + * to be set again or the next flush is skipped. + */ + private void markCurrentFileDirty() { + if (data.writer == null) { + return; + } + TextFileOutputData.IFileStreamsCollection coll = data.getFileStreamsCollection(); + if (coll == null) { + return; + } + TextFileOutputData.FileStream last = coll.getLastStream(); + if (last != null && last.getBufferedOutputStream() == data.writer) { + last.setDirty(true); + return; + } + coll.forEachOpenStream( + stream -> { + if (stream.getBufferedOutputStream() == data.writer) { + stream.setDirty(true); + } + }); + } + public String buildFilename(String filename, boolean ziparchive) { return meta.buildFilename( filename, diff --git a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java index 22b06ea1036..810f23794bc 100644 --- a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java +++ b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java @@ -126,13 +126,18 @@ public void setDirty(boolean dirty) { } public void flush() throws IOException { - if (isDirty) { + if (isDirty && getBufferedOutputStream() != null) { getBufferedOutputStream().flush(); isDirty = false; } } public void close() throws IOException { + // Flush even when the dirty flag was cleared. Otherwise bytes written after the last + // interval flush are dropped when the buffer is discarded below. + if (getBufferedOutputStream() != null) { + getBufferedOutputStream().flush(); + } setBufferedOutputStream(null); getCompressedOutputStream().close(); setCompressedOutputStream(null); diff --git a/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java b/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java new file mode 100644 index 00000000000..bf63a2cbaac --- /dev/null +++ b/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hop.pipeline.transforms.textfileoutput; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.ByteArrayOutputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import org.apache.hop.core.Const; +import org.apache.hop.core.compress.CompressionPluginType; +import org.apache.hop.core.plugins.PluginRegistry; +import org.apache.hop.core.row.IRowMeta; +import org.apache.hop.core.row.RowMeta; +import org.apache.hop.core.row.value.ValueMetaString; +import org.apache.hop.core.variables.IVariables; +import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension; +import org.apache.hop.pipeline.transforms.mock.TransformMockHelper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.mockito.Mockito; + +class TextFileOutputFlushTest { + @RegisterExtension + static RestoreHopEngineEnvironmentExtension env = new RestoreHopEngineEnvironmentExtension(); + + private TransformMockHelper helper; + + @BeforeAll + static void setUpBeforeClass() throws Exception { + PluginRegistry.addPluginType(CompressionPluginType.getInstance()); + PluginRegistry.init(); + } + + @BeforeEach + void setUp() { + helper = + new TransformMockHelper<>( + "Text file output flush", TextFileOutputMeta.class, TextFileOutputData.class); + Mockito.when(helper.logChannelFactory.create(Mockito.any(), Mockito.any())) + .thenReturn(helper.iLogChannel); + Mockito.when(helper.pipeline.isRunning()).thenReturn(true); + } + + @AfterEach + void tearDown() { + helper.cleanUp(); + } + + @Test + void flushIntervalDefaultsToAFewSeconds() throws Exception { + FlushProbe transform = newProbe(); + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, null); + assertEquals(TextFileOutput.DEFAULT_FILE_FLUSH_INTERVAL_MS, transform.getFlushInterval()); + + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "0"); + assertEquals(TextFileOutput.DEFAULT_FILE_FLUSH_INTERVAL_MS, transform.getFlushInterval()); + + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "nope"); + assertEquals(TextFileOutput.DEFAULT_FILE_FLUSH_INTERVAL_MS, transform.getFlushInterval()); + + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "2500"); + assertEquals(2500, transform.getFlushInterval()); + } + + @Test + void slowRowsAreFlushedOnTheIntervalAndAgainAfterThat() throws Exception { + FlushProbe transform = newProbe(); + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "1000"); + assertTrue(transform.init()); + + transform.now = 1_000_000L; + transform.row = new Object[] {"a"}; + assertTrue(transform.processRow()); + assertEquals("", transform.written(), "first row stays buffered until the interval elapses"); + + transform.now = 1_000_500L; + transform.row = new Object[] {"b"}; + assertTrue(transform.processRow()); + assertEquals("", transform.written(), "a row inside the interval does not flush"); + + transform.now = 1_001_001L; + transform.row = new Object[] {"c"}; + assertTrue(transform.processRow()); + assertEquals("a\nb\nc\n", transform.written()); + + transform.now = 1_002_002L; + transform.row = new Object[] {"d"}; + assertTrue(transform.processRow()); + assertEquals("a\nb\nc\nd\n", transform.written()); + } + + private FlushProbe newProbe() { + TextFileOutputMeta meta = new TextFileOutputMeta(); + meta.setDefault(); + meta.setHeaderEnabled(false); + meta.setFooterEnabled(false); + meta.setSeparator(""); + meta.setEnclosure(""); + meta.setFileFormat("UNIX"); + meta.setEncoding(Const.UTF_8); + meta.setCreateParentFolder(false); + meta.setEndedLine(null); + meta.getFileSettings().setFileName("out.txt"); + meta.getFileSettings().setExtension(""); + meta.getFileSettings().setDoNotOpenNewFileInit(true); + meta.getFileSettings().setAddToResultFiles(false); + meta.getFileSettings().setFastDump(true); + + FlushProbe transform = + new FlushProbe( + helper.transformMeta, + meta, + new TextFileOutputData(), + helper.pipelineMeta, + helper.pipeline); + RowMeta rowMeta = new RowMeta(); + ValueMetaString column = new ValueMetaString("col"); + column.setLength(-1); + rowMeta.addValueMeta(column); + transform.setInputRowMeta(rowMeta); + return transform; + } + + private static final class FlushProbe extends TextFileOutput { + private long now; + private Object[] row; + private final ByteArrayOutputStream written = new ByteArrayOutputStream(); + + private FlushProbe( + org.apache.hop.pipeline.transform.TransformMeta transformMeta, + TextFileOutputMeta meta, + TextFileOutputData data, + org.apache.hop.pipeline.PipelineMeta pipelineMeta, + org.apache.hop.pipeline.Pipeline pipeline) { + super(transformMeta, meta, data, 0, pipelineMeta, pipeline); + } + + private String written() { + return written.toString(StandardCharsets.UTF_8); + } + + @Override + protected long currentFlushTimeMillis() { + return now; + } + + @Override + public Object[] getRow() { + return row; + } + + @Override + public void putRow(IRowMeta rowMeta, Object[] row) { + // The flush test does not chain rows to a downstream transform. + } + + @Override + protected OutputStream getOutputStream( + String vfsFilename, IVariables variables, boolean append) { + return new OutputStream() { + @Override + public void write(int b) { + written.write(b); + } + + @Override + public void write(byte[] b, int off, int len) { + written.write(b, off, len); + } + }; + } + } +} From 6d56bd1ffe82c999a212d5a61f7dae00294bfbf9 Mon Sep 17 00:00:00 2001 From: mattcasters Date: Thu, 1 Oct 2026 19:58:56 +0200 Subject: [PATCH 2/2] Issue #2406 : Close text files after an interval flush and let -1 disable it --- .../main/java/org/apache/hop/core/Const.java | 5 +- .../pipeline/transforms/textfileoutput.adoc | 2 +- .../modules/ROOT/pages/variables.adoc | 1 + .../textfileoutput/TextFileOutput.java | 52 ++++---- .../textfileoutput/TextFileOutputData.java | 39 +++--- .../TextFileOutputFlushTest.java | 116 ++++++++++++++++++ 6 files changed, 167 insertions(+), 48 deletions(-) diff --git a/core/src/main/java/org/apache/hop/core/Const.java b/core/src/main/java/org/apache/hop/core/Const.java index 5c07ffa830e..bb0bb4d1e84 100644 --- a/core/src/main/java/org/apache/hop/core/Const.java +++ b/core/src/main/java/org/apache/hop/core/Const.java @@ -812,12 +812,13 @@ public static boolean toBoolean(String string) { /** * Milliseconds between flushes of all open files in the Text File Output transform. {@code 0} - * selects the transform default of 5000. + * selects the transform default of 5000. A negative value, for example {@code -1}, disables the + * interval flush. */ @Variable( value = "5000", description = - "This project variable is used by the Text File Output transform. It defines how many milliseconds to wait between flushes of files opened by the transform. Output is buffered, so slow input stays invisible until the buffer fills or the file is closed. The default is 5000 (5 seconds). A value of 0 uses that default. Set a positive number of milliseconds to change the interval.") + "This project variable is used by the Text File Output transform. It defines how many milliseconds to wait between flushes of files opened by the transform. Output is buffered, so slow input stays invisible until the buffer fills or the file is closed. The default is 5000 (5 seconds). A value of 0 uses that default. Set a positive number of milliseconds to change the interval. A negative value, for example -1, disables the interval flush.") public static final String HOP_FILE_OUTPUT_MAX_STREAM_LIFE = "HOP_FILE_OUTPUT_MAX_STREAM_LIFE"; /** Set this variable to Y to disable standard Hop logging to the console. (stdout) */ diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc index 69e938aa04f..45f66ce1619 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/textfileoutput.adoc @@ -29,7 +29,7 @@ This is commonly used to generate Comma Separated Values (CSV files) that can be It is also possible to generate fixed width files by setting lengths on the fields in the fields tab. -Output is buffered. Open files are flushed on an interval so slow-arriving rows show up without waiting for that buffer to fill. The interval is xref:variables.adoc[HOP_FILE_OUTPUT_MAX_STREAM_LIFE], in milliseconds. The default is 5 seconds. A value of 0 uses that default. Set a positive number of milliseconds to change the interval. +Output is buffered. Open files are flushed on an interval so slow-arriving rows show up without waiting for that buffer to fill. The interval is xref:variables.adoc[HOP_FILE_OUTPUT_MAX_STREAM_LIFE], in milliseconds. The default is 5 seconds. A value of 0 uses that default. Set a positive number of milliseconds to change the interval. A negative value, for example -1, disables the interval flush. You can choose to use a xref:metadata-types/static-schema-definition.adoc[Schema Definition] or to define the required fields' layout manually. If you decide to define the fields layout by using a xref:metadata-types/static-schema-definition.adoc[Schema Definition], use the xref:pipeline/transforms/schemamapping.adoc[Schema mapping] transform to adjust the incoming stream according to the chosen xref:metadata-types/static-schema-definition.adoc[Schema Definition] diff --git a/docs/hop-user-manual/modules/ROOT/pages/variables.adoc b/docs/hop-user-manual/modules/ROOT/pages/variables.adoc index 37ec6165181..e0b6af0efbb 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/variables.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/variables.adoc @@ -264,6 +264,7 @@ Output is buffered, so slow input stays invisible until the buffer fills or the The default is 5000 (5 seconds). A value of 0 uses that default. Set a positive number of milliseconds to change the interval. +A negative value, for example -1, disables the interval flush. |HOP_GLOBAL_LOG_VARIABLES_CLEAR_ON_EXPORT|N|Set this variable to N to preserve global log variables defined in pipeline / workflow Properties -> Log panel. Changing it to true will clear it when export pipeline / workflow. |HOP_JSON_INPUT_INCLUDE_NULLS|Y|Name of te variable to set so that Nulls are considered while parsing JSON files. If HOP_JSON_INPUT_INCLUDE_NULLS is "Y" then nulls will be included otherwise they will not be included (default behavior) diff --git a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java index 78e4ca2e1d9..002f981aac5 100644 --- a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java +++ b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutput.java @@ -123,7 +123,7 @@ private ICompressionProvider getCompressionProvider() throws HopException { @SuppressWarnings("java:S2095") // the stream is owned by the transform and closed in closeFile() public void initFileStreamWriter(String filename) throws HopException { - data.writer = null; + assignWriter(null, null); try { TextFileOutputData.FileStream fileStreams = null; @@ -247,7 +247,7 @@ public void initFileStreamWriter(String filename) throws HopException { data.fos = fileStreams.getFileOutputStream(); data.out = fileStreams.getCompressedOutputStream(); - data.writer = fileStreams.getBufferedOutputStream(); + assignWriter(fileStreams.getBufferedOutputStream(), fileStreams); } catch (HopException ke) { throw ke; } catch (Exception e) { @@ -289,17 +289,18 @@ public String getOutputFileName(Object[] row) throws HopException { } /** - * Milliseconds between flushes when {@link Const#HOP_FILE_OUTPUT_MAX_STREAM_LIFE} is unset or not - * a positive number. A few seconds, so slow input shows up without waiting for the buffer to - * fill. + * Milliseconds between flushes when {@link Const#HOP_FILE_OUTPUT_MAX_STREAM_LIFE} is unset, not a + * number, or {@code 0}. A few seconds, so slow input shows up without waiting for the buffer to + * fill. A negative value disables the interval flush. */ static final int DEFAULT_FILE_FLUSH_INTERVAL_MS = 5000; public int getFlushInterval() { String maxStreamLife = variables.getVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE); int flushInterval = Const.toInt(maxStreamLife, DEFAULT_FILE_FLUSH_INTERVAL_MS); - // 0 is the historical default and means "not configured". - if (flushInterval <= 0) { + // 0 is what existing hop-config.json files store and means "not configured". + // Only a negative value disables the interval flush. + if (flushInterval == 0) { return DEFAULT_FILE_FLUSH_INTERVAL_MS; } return flushInterval; @@ -373,7 +374,7 @@ private boolean writeRowToFile(Object[] row) throws HopException { data.splitnr++; data.fos = null; data.out = null; - data.writer = null; + assignWriter(null, null); filename = getOutputFileName(null); isWriteHeader = isWriteHeader(filename); initFileStreamWriter(filename); @@ -824,29 +825,22 @@ && containsSeparatorOrEnclosure( return retval; } + /** + * Writer and the stream it belongs to move together so dirty-marking does not scan open files. + */ + private void assignWriter(OutputStream writer, TextFileOutputData.FileStream fileStream) { + data.writer = writer; + data.currentFileStream = fileStream; + } + /** * An interval flush clears the dirty flag. Later rows still land in the buffer, so the flag has * to be set again or the next flush is skipped. */ private void markCurrentFileDirty() { - if (data.writer == null) { - return; + if (data.currentFileStream != null) { + data.currentFileStream.setDirty(true); } - TextFileOutputData.IFileStreamsCollection coll = data.getFileStreamsCollection(); - if (coll == null) { - return; - } - TextFileOutputData.FileStream last = coll.getLastStream(); - if (last != null && last.getBufferedOutputStream() == data.writer) { - last.setDirty(true); - return; - } - coll.forEachOpenStream( - stream -> { - if (stream.getBufferedOutputStream() == data.writer) { - stream.setDirty(true); - } - }); } public String buildFilename(String filename, boolean ziparchive) { @@ -1000,7 +994,7 @@ protected boolean closeFile() { coll.closeStream(data.writer); } } - data.writer = null; + assignWriter(null, null); data.out = null; data.fos = null; if (isDebug()) { @@ -1010,7 +1004,7 @@ protected boolean closeFile() { } catch (Exception e) { logError("Exception trying to close file: " + e.toString()); setErrors(1); - data.writer = null; + assignWriter(null, null); data.out = null; data.fos = null; retval = false; @@ -1148,7 +1142,7 @@ protected void close() throws IOException { } coll.flushOpenFiles(true); } - data.writer = null; + assignWriter(null, null); } private void emitWriteLineageForOpenStream(String filename, TextFileOutputData.FileStream fs) { @@ -1186,7 +1180,7 @@ public void dispose() { logError("Unexpected error closing file", e); setErrors(1); } - data.writer = null; + assignWriter(null, null); data.out = null; data.fos = null; diff --git a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java index 810f23794bc..70c2ac50fc8 100644 --- a/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java +++ b/plugins/transforms/textfile/src/main/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputData.java @@ -213,16 +213,17 @@ public void closeOldestOpenFile(boolean removeFileFromCollection) throws IOExcep @Override public void flushOpenFiles(boolean closeAfterFlush) throws IOException { for (FileStream outputStream : streamsList) { - if (outputStream.isDirty()) { - try { + try { + if (outputStream.isDirty()) { outputStream.flush(); - if (closeAfterFlush && outputStream.isOpen()) { - outputStream.close(); - numOpenFiles--; - } - } catch (IOException e) { - e.printStackTrace(); } + // An interval flush clears the dirty flag. End-of-run still has to close the stream. + if (closeAfterFlush && outputStream.isOpen()) { + outputStream.close(); + numOpenFiles--; + } + } catch (IOException e) { + e.printStackTrace(); } } } @@ -405,15 +406,18 @@ public void closeOldestOpenFile(boolean removeFileFromCollection) throws IOExcep @Override public void flushOpenFiles(boolean closeAfterFlush) { for (FileStreamsCollectionEntry collectionEntry : indexMap.values()) { - if (collectionEntry.getFileStream().isDirty()) { - try { - collectionEntry.getFileStream().flush(); - if (closeAfterFlush) { - collectionEntry.getFileStream().close(); - } - } catch (IOException e) { - e.printStackTrace(); + FileStream fileStream = collectionEntry.getFileStream(); + try { + if (fileStream.isDirty()) { + fileStream.flush(); + } + // An interval flush clears the dirty flag. End-of-run still has to close the stream. + if (closeAfterFlush && fileStream.isOpen()) { + fileStream.close(); + numOpenFiles--; } + } catch (IOException e) { + e.printStackTrace(); } } } @@ -456,6 +460,9 @@ public void closeStream(OutputStream outputStream) throws IOException { public OutputStream writer; + /** Stream {@link #writer} currently writes to, so a row can mark that file dirty directly. */ + public FileStream currentFileStream; + public DecimalFormat defaultDecimalFormat; public DecimalFormatSymbols defaultDecimalFormatSymbols; diff --git a/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java b/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java index bf63a2cbaac..69063716438 100644 --- a/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java +++ b/plugins/transforms/textfile/src/test/java/org/apache/hop/pipeline/transforms/textfileoutput/TextFileOutputFlushTest.java @@ -18,13 +18,20 @@ package org.apache.hop.pipeline.transforms.textfileoutput; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.io.BufferedOutputStream; +import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.io.OutputStream; import java.nio.charset.StandardCharsets; +import java.util.zip.GZIPInputStream; import org.apache.hop.core.Const; +import org.apache.hop.core.compress.CompressionOutputStream; import org.apache.hop.core.compress.CompressionPluginType; +import org.apache.hop.core.compress.gzip.GzipCompressionProvider; import org.apache.hop.core.plugins.PluginRegistry; import org.apache.hop.core.row.IRowMeta; import org.apache.hop.core.row.RowMeta; @@ -80,6 +87,86 @@ void flushIntervalDefaultsToAFewSeconds() throws Exception { transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "2500"); assertEquals(2500, transform.getFlushInterval()); + + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "-1"); + assertEquals(-1, transform.getFlushInterval()); + } + + @Test + void negativeIntervalDoesNotFlushOnTheClock() throws Exception { + FlushProbe transform = newProbe(); + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "-1"); + assertTrue(transform.init()); + + transform.now = 1_000_000L; + transform.row = new Object[] {"a"}; + assertTrue(transform.processRow()); + + transform.now = 1_060_000L; + transform.row = new Object[] {"b"}; + assertTrue(transform.processRow()); + assertEquals("", transform.written(), "a negative interval does not flush on the clock"); + } + + @Test + void intervalFlushAfterLastRowStillClosesGzipFile() throws Exception { + FlushProbe transform = newProbe("GZip"); + transform.setVariable(Const.HOP_FILE_OUTPUT_MAX_STREAM_LIFE, "1000"); + assertTrue(transform.init()); + + transform.now = 1_000_000L; + transform.row = new Object[] {"a"}; + assertTrue(transform.processRow()); + + // Last data row. The interval elapses while writing it, so the flush clears the dirty flag. + transform.now = 1_002_000L; + transform.row = new Object[] {"b"}; + assertTrue(transform.processRow()); + assertFalse(transform.currentStreamDirty(), "interval flush cleared the dirty flag"); + assertTrue(transform.currentStreamOpen(), "interval flush must not close the file"); + assertFalse(transform.isOutputClosed()); + + transform.row = null; + assertFalse(transform.processRow()); + assertFalse(transform.currentStreamOpen()); + assertTrue(transform.isOutputClosed()); + assertEquals("a\nb\n", gunzip(transform.gzipBytes())); + } + + @Test + void closeAfterFlushClosesStreamThatIntervalFlushAlreadyCleaned() throws Exception { + TextFileOutputData data = new TextFileOutputData(); + assertCleanOpenStreamIsClosed(data.new FileStreamsList()); + assertCleanOpenStreamIsClosed(data.new FileStreamsMap()); + } + + private static void assertCleanOpenStreamIsClosed(TextFileOutputData.IFileStreamsCollection coll) + throws Exception { + String collection = coll.getClass().getSimpleName(); + ByteArrayOutputStream raw = new ByteArrayOutputStream(); + CompressionOutputStream compression = new GzipCompressionProvider().createOutputStream(raw); + BufferedOutputStream buffered = new BufferedOutputStream(compression, 5000); + TextFileOutputData.FileStream stream = + new TextFileOutputData().new FileStream(raw, compression, buffered); + buffered.write("hello\n".getBytes(StandardCharsets.UTF_8)); + stream.setDirty(true); + coll.add("out.txt", stream); + + coll.flushOpenFiles(false); + assertFalse(stream.isDirty(), collection); + assertTrue(stream.isOpen(), collection); + assertEquals(1, coll.getNumOpenFiles(), collection); + + coll.flushOpenFiles(true); + assertFalse(stream.isOpen(), collection); + assertEquals(0, coll.getNumOpenFiles(), collection); + assertEquals("hello\n", gunzip(raw.toByteArray()), collection); + } + + private static String gunzip(byte[] gzipBytes) throws IOException { + try (GZIPInputStream in = new GZIPInputStream(new ByteArrayInputStream(gzipBytes))) { + return new String(in.readAllBytes(), StandardCharsets.UTF_8); + } } @Test @@ -110,8 +197,13 @@ void slowRowsAreFlushedOnTheIntervalAndAgainAfterThat() throws Exception { } private FlushProbe newProbe() { + return newProbe("None"); + } + + private FlushProbe newProbe(String compression) { TextFileOutputMeta meta = new TextFileOutputMeta(); meta.setDefault(); + meta.setFileCompression(compression); meta.setHeaderEnabled(false); meta.setFooterEnabled(false); meta.setSeparator(""); @@ -144,6 +236,7 @@ private FlushProbe newProbe() { private static final class FlushProbe extends TextFileOutput { private long now; private Object[] row; + private boolean outputClosed; private final ByteArrayOutputStream written = new ByteArrayOutputStream(); private FlushProbe( @@ -159,6 +252,24 @@ private String written() { return written.toString(StandardCharsets.UTF_8); } + private byte[] gzipBytes() { + return written.toByteArray(); + } + + private boolean isOutputClosed() { + return outputClosed; + } + + private boolean currentStreamOpen() { + TextFileOutputData.FileStream last = data.getFileStreamsCollection().getLastStream(); + return last != null && last.isOpen(); + } + + private boolean currentStreamDirty() { + TextFileOutputData.FileStream last = data.getFileStreamsCollection().getLastStream(); + return last != null && last.isDirty(); + } + @Override protected long currentFlushTimeMillis() { return now; @@ -187,6 +298,11 @@ public void write(int b) { public void write(byte[] b, int off, int len) { written.write(b, off, len); } + + @Override + public void close() { + outputClosed = true; + } }; } }