do not abort instant when some write task was failed

apache · Jun 24, 2022 · 6099161 · 6099161
1 parent 237c256
commit 6099161
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 4 deletions.
diff --git a/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java b/...asource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
@@ -266,9 +266,12 @@ public void notifyCheckpointComplete(long checkpointId) {
   @Override
   public void notifyCheckpointAborted(long checkpointId) {
     if (checkpointId == this.checkpointId) {
-      executor.execute(() -> {
-        this.ckpMetadata.abortInstant(this.instant);
-      }, "abort instant %s", this.instant);
+      // write task failed we do not reuse the instant
+      if (Arrays.stream(eventBuffer).anyMatch(s -> s == null)) {
+        executor.execute(() -> {
+          this.ckpMetadata.abortInstant(this.instant);
+        }, "abort instant %s", this.instant);
+      }
     }
   }
 

diff --git a/...rce/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java b/...rce/hudi-flink/src/main/java/org/apache/hudi/sink/common/AbstractStreamWriteFunction.java
@@ -265,6 +265,6 @@ protected String instantToWrite(boolean hasData) {
    * Returns whether the pending instant is invalid to write with.
    */
   private boolean invalidInstant(String instant, boolean hasData) {
-    return instant.equals(this.currentInstant) && hasData && this.ckpMetadata.isAborted(instant);
+    return instant.equals(this.currentInstant) && hasData && !this.ckpMetadata.isAborted(instant);
   }
 }