rollback task failed scene

apache · Jun 27, 2022 · 3e8a36c · 3e8a36c
1 parent 1b1334c
commit 3e8a36c
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 12 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
@@ -50,7 +50,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.Collection;
@@ -266,7 +265,7 @@ public void notifyCheckpointComplete(long checkpointId) {
 
   @Override
   public void notifyCheckpointAborted(long checkpointId) {
-    if (checkpointId == this.checkpointId) {
+    if (checkpointId == this.checkpointId && !WriteMetadataEvent.BOOTSTRAP_INSTANT.equals(this.instant)) {
       executor.execute(() -> {
         this.ckpMetadata.abortInstant(this.instant);
       }, "abort instant %s", this.instant);
@@ -306,13 +305,6 @@ public void subtaskFailed(int i, @Nullable Throwable throwable) {
     // reset the event
     this.eventBuffer[i] = null;
     LOG.warn("Reset the event for task [" + i + "]", throwable);
-    if (Arrays.stream(this.eventBuffer).allMatch(event -> event == null)) {
-      try {
-        this.ckpMetadata.bootstrap(this.metaClient);
-      } catch (IOException e) {
-        throw new HoodieException("Bootstrap ckpMetadata exception", e);
-      }
-    }
   }
 
   @Override

diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/meta/CkpMetadata.java
@@ -152,9 +152,7 @@ public void commitInstant(String instant) {
   public void abortInstant(String instant) {
     Path path = fullPath(CkpMessage.getFileName(instant, CkpMessage.State.ABORTED));
     try {
-      if (fs.exists(fullPath(CkpMessage.getFileName(instant, CkpMessage.State.INFLIGHT)))) {
-        fs.createNewFile(path);
-      }
+      fs.createNewFile(path);
     } catch (IOException e) {
       throw new HoodieException("Exception while adding checkpoint abort metadata for instant: " + instant);
     }