Issue #2406 : Flush text file output every few seconds by default

core/src/main/java/org/apache/hop/core/Const.java

@@ -811,13 +811,14 @@
   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. A negative value, for example {@code -1}, disables the
+   * interval flush.
    */
   @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. 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) */
@@ -1218,7 +1219,7 @@
    * @return The rounded floating point value
    */
   @Deprecated(since = "2.10")
   @SuppressWarnings("java:S1874") // Ignore BigDecimal RoundingModes as code will be removed
   public static BigDecimal round(BigDecimal f, int places, int roundingMode) {
     if (roundingMode == ROUND_HALF_CEILING) {
       if (f.signum() >= 0) {
@@ -2021,7 +2022,7 @@
         BufferedReader br;
         try {
           Process pr = Runtime.getRuntime().exec("hostname");
           br = new BufferedReader(new InputStreamReader(pr.getInputStream()));
           String line;
           if ((line = br.readLine()) != null) {
             return line;
@@ -2087,7 +2088,7 @@
     String s = "";
     boolean errorOccured = false;
     if (os.equalsIgnoreCase("Windows NT")
         || os.equalsIgnoreCase("Windows 2000")
         || os.equalsIgnoreCase("Windows XP")
         || os.equalsIgnoreCase("Windows 95")
         || os.equalsIgnoreCase("Windows 98")

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. 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]
 
 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].

docs/hop-user-manual/modules/ROOT/pages/variables.adoc

@@ -258,8 +258,13 @@ 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.
+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)

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);

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++;
+    }
+  }
 }

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;
@@ -124,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;
 
@@ -248,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,15 +288,20 @@ public String getOutputFileName(Object[] row) throws HopException {
     return filename;
   }
 
+  /**
+   * 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("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 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;
   }
@@ -370,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);
@@ -388,16 +392,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 +456,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 +534,7 @@ public void writeRow(IRowMeta rowMeta, Object[] r) throws HopTransformException
       }
 
       incrementLinesOutput();
+      markCurrentFileDirty();
 
     } catch (Exception e) {
       throw new HopTransformException("Error writing line", e);
@@ -732,6 +742,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 +821,28 @@ && containsSeparatorOrEnclosure(
       retval = true;
     }
     incrementLinesOutput();
+    markCurrentFileDirty();
     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.currentFileStream != null) {
+      data.currentFileStream.setDirty(true);
+    }
+  }
+
   public String buildFilename(String filename, boolean ziparchive) {
     return meta.buildFilename(
         filename,
@@ -964,7 +994,7 @@ protected boolean closeFile() {
           coll.closeStream(data.writer);
         }
       }
-      data.writer = null;
+      assignWriter(null, null);
       data.out = null;
       data.fos = null;
       if (isDebug()) {
@@ -974,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;
@@ -1112,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) {
@@ -1150,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;
 

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);
@@ -208,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();
         }
       }
     }
@@ -400,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();
         }
       }
     }
@@ -451,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;