[HUDI-7109] Fix Flink may re-use a committed instant in append mode (a…

…pache#10119)
jonvex · Nov 17, 2023 · 3d0c450 · 3d0c450
1 parent f06ff5b
commit 3d0c450
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunction.java b/...-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/append/AppendWriteFunction.java
@@ -112,7 +112,7 @@ protected void sendBootstrapEvent() {
     int attemptId = getRuntimeContext().getAttemptNumber();
     if (attemptId > 0) {
       // either a partial or global failover, reuses the current inflight instant
-      if (this.currentInstant != null) {
+      if (this.currentInstant != null && !metaClient.getActiveTimeline().filterCompletedInstants().containsInstant(currentInstant)) {
         LOG.info("Recover task[{}] for instant [{}] with attemptId [{}]", taskID, this.currentInstant, attemptId);
         this.currentInstant = null;
         return;