(beam) branch master updated: Avoid excessive flushing in DataflowWorkerLoggingHandler. (#31175)

[yh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new f468d40856d Avoid excessive flushing in DataflowWorkerLoggingHandler. (#31175)
f468d40856d is described below

commit f468d40856d32286e4bd435e73a212f1fb25e116
Author: Nick Anikin <52...@users.noreply.github.com>
AuthorDate: Fri May 3 12:57:58 2024 -0700

    Avoid excessive flushing in DataflowWorkerLoggingHandler. (#31175)
    
    * Avoid excessive flushing in DataflowWorkerLoggingHandler.
    
    This is a continuation of https://github.com/apache/beam/pull/25922 and an attempt to fix an issue with Dataflow logging, when log messages sometimes get corrupted.
    
    DataflowWorkerLoggingHandler logs message to a physical file, then a separate process in Dataflow reads this file and exports logs to Cloud Logging.
    
    I found that the Jackson generator used by DataflowWorkerLoggingHandler has the SerializationFeature.FLUSH_AFTER_WRITE_VALUE feature enabled by default, which makes it flush the JSON output to the underlying stream after every object write. DataflowWorkerLoggingHandler currently writes the "severity" field as an object (generator.writeObjectField("severity", ...))., which results in the log message being flushed between "severity" and "message" fields, and some time Dataflow logging i [...]
    
    Disabling SerializationFeature.FLUSH_AFTER_WRITE_VALUE should fix this issue (and improve logging performance a little bit).
    
    * Fix formatting.
---
 .../dataflow/worker/logging/DataflowWorkerLoggingHandler.java    | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
index 44daec2379a..572f9354ca9 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.util.MinimalPrettyPrinter;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
 import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
@@ -111,7 +112,13 @@ public class DataflowWorkerLoggingHandler extends Handler {
   DataflowWorkerLoggingHandler(Supplier<OutputStream> factory, long sizeLimit) throws IOException {
     this.setFormatter(new SimpleFormatter());
     this.outputStreamFactory = factory;
-    this.generatorFactory = new ObjectMapper().getFactory();
+    this.generatorFactory =
+        new ObjectMapper()
+            // Required to avoid flushing to the file in the middle of a log message and potentially
+            // breaking its JSON formatting, if the first part is read from the file before the rest
+            // of the message:
+            .disable(SerializationFeature.FLUSH_AFTER_WRITE_VALUE)
+            .getFactory();
     this.sizeLimit = sizeLimit < 1 ? Long.MAX_VALUE : sizeLimit;
     createOutputStream();
   }