feat: add record write failure log and metrics (#13417)

Author: HuangZhenQiu

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/WriteStatus.java

@@ -91,7 +91,7 @@ public WriteStatus(Boolean trackSuccessRecords, Double failureFraction) {
   public WriteStatus() {
     this.failureFraction = 0.0d;
     this.trackSuccessRecords = false;
-    this.random = null;
+    this.random = new Random(RANDOM_SEED);
     this.isMetadataTable = false;
   }
 

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkStreamWriteMetrics.java

@@ -66,6 +66,11 @@ public class FlinkStreamWriteMetrics extends HoodieFlinkMetrics {
    */
   private long numOfFilesWritten;
 
+  /**
+   * Number of record write failure during a checkpoint window.
+   */
+  private long numOfRecordWriteFailures;
+
   /**
    * Number of records written per seconds.
    */
@@ -104,6 +109,8 @@ public void registerMetrics() {
     metricGroup.gauge("fileFlushTotalCosts", () -> fileFlushTotalCosts);
     metricGroup.gauge("numOfFilesWritten", () -> numOfFilesWritten);
     metricGroup.gauge("numOfOpenHandle", () -> numOfOpenHandle);
+    metricGroup.gauge("numOfRecordWriteFailures", () -> numOfRecordWriteFailures);
+
 
     metricGroup.meter("handleSwitchPerSecond", handleSwitchPerSecond);
 
@@ -132,6 +139,14 @@ public void increaseNumOfFilesWritten() {
     numOfFilesWritten += 1;
   }
 
+  public void increaseNumOfRecordWriteFailure(long recordWriteFailures) {
+    numOfRecordWriteFailures += recordWriteFailures;
+  }
+
+  public long getNumOfRecordWriteFailures() {
+    return numOfRecordWriteFailures;
+  }
+
   public void increaseNumOfOpenHandle() {
     numOfOpenHandle += 1;
     increaseNumOfFilesWritten();
@@ -165,6 +180,7 @@ public void resetAfterCommit() {
     this.numOfOpenHandle = 0;
     this.writeBufferedSize = 0;
     this.fileFlushTotalCosts = 0;
+    this.numOfRecordWriteFailures = 0;
   }
 
 }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunction.java

@@ -19,6 +19,8 @@
 package org.apache.hudi.sink.append;
 
 import org.apache.hudi.client.WriteStatus;
+
+import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.metrics.FlinkStreamWriteMetrics;
@@ -40,6 +42,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Sink function to write the data to the underneath filesystem.
@@ -144,8 +147,8 @@ private void flushData(boolean endInput) {
       LOG.info("No data to write in subtask [{}] for instant [{}]", taskID, this.currentInstant);
     }
 
+    recordWriteFailure(writeMetrics, writeStatus);
     StreamerUtil.validateWriteStatus(config, currentInstant, writeStatus);
-
     final WriteMetadataEvent event = WriteMetadataEvent.builder()
         .taskID(taskID)
         .checkpointId(this.checkpointId)
@@ -167,4 +170,26 @@ private void registerMetrics() {
     writeMetrics = new FlinkStreamWriteMetrics(metrics);
     writeMetrics.registerMetrics();
   }
+
+  /**
+   * Update metrics and log for errors in write status.
+   *
+   * @param writeMetrics FlinkStreamWriteMetrics
+   * @param writeStatus write status from write handler
+   */
+  @VisibleForTesting
+  static void recordWriteFailure(FlinkStreamWriteMetrics writeMetrics, List<WriteStatus> writeStatus) {
+    Map.Entry<HoodieKey, Throwable> firstFailure = null;
+    for (WriteStatus status : writeStatus) {
+      writeMetrics.increaseNumOfRecordWriteFailure(status.getTotalErrorRecords());
+      if (firstFailure == null && status.getErrors().size() > 0) {
+        firstFailure = status.getErrors().entrySet().stream().findFirst().get();
+      }
+    }
+
+    // Only print the first record failure to prevent logs occupy too much disk in worst case.
+    if (firstFailure != null) {
+      LOG.error("The first record with written failure {}", firstFailure.getKey().getRecordKey(), firstFailure.getValue());
+    }
+  }
 }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/append/TestAppendWriteFunction.java

@@ -31,12 +31,30 @@
 import java.util.List;
 
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+
+import org.apache.hudi.metrics.FlinkStreamWriteMetrics;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
 
 /**
  * Test cases for {@link AppendWriteFunction}.
  */
 public class TestAppendWriteFunction {
 
+  private FlinkStreamWriteMetrics flinkStreamWriteMetrics;
+  private UnregisteredMetricsGroup metricGroup;
+
+  @BeforeEach
+  void setUp() {
+    metricGroup = new UnregisteredMetricsGroup();
+    flinkStreamWriteMetrics = new FlinkStreamWriteMetrics(metricGroup);
+  }
+
   @Test
   void testRecordWriteNoFailure() {
     WriteStatus writeStatus = new WriteStatus();
@@ -61,7 +79,7 @@ void testRecordWriteFailureValidationWithoutFailFast() {
   void testRecordWriteFailureValidationWithFailFast() {
     WriteStatus writeStatus = new WriteStatus();
     writeStatus.markFailure(
-            "key1", "/partition1", new RuntimeException("test exception"));
+        "key1", "/partition1", new RuntimeException("test exception"));
     List<WriteStatus> writeStatusList = Collections.singletonList(writeStatus);
 
     Configuration configuration = new Configuration();
@@ -71,4 +89,36 @@ void testRecordWriteFailureValidationWithFailFast() {
     assertThrows(HoodieException.class,
         () -> StreamerUtil.validateWriteStatus(configuration, HoodieInstantTimeGenerator.getCurrentInstantTimeStr(), writeStatusList));
   }
+
+  @Test
+  void testRecordWriteFailure() {
+    WriteStatus writeStatus = new WriteStatus();
+    writeStatus.markFailure(
+        "key1", "/partition1", new RuntimeException("test exception"));
+    List<WriteStatus> writeStatusList = Arrays.asList(writeStatus);
+
+    AppendWriteFunction.recordWriteFailure(flinkStreamWriteMetrics, writeStatusList);
+
+    // Verify that the failure was recorded in metrics
+    assertEquals(1, flinkStreamWriteMetrics.getNumOfRecordWriteFailures());
+  }
+
+  @Test
+  void testRecordWriteFailureMultipleErrors() {
+    WriteStatus writeStatus1 = new WriteStatus();
+    writeStatus1.markFailure("key1", "/partition1", new RuntimeException("error 1"));
+    writeStatus1.markFailure("key2", "/partition1", new RuntimeException("error 2"));
+    writeStatus1.markFailure("key3", "/partition1", new RuntimeException("error 3"));
+
+    WriteStatus writeStatus2 = new WriteStatus();
+    writeStatus2.markFailure("key4", "/partition2", new IllegalArgumentException("illegal argument"));
+    writeStatus2.markFailure("key5", "/partition2", new NullPointerException("null pointer"));
+
+    List<WriteStatus> writeStatusList = Arrays.asList(writeStatus1, writeStatus2);
+
+    AppendWriteFunction.recordWriteFailure(flinkStreamWriteMetrics, writeStatusList);
+
+    // Should record total 5 failures across both write statuses
+    assertEquals(5, flinkStreamWriteMetrics.getNumOfRecordWriteFailures());
+  }
 }