[FLINK-26801] Support duplicate checkpoint aborted messages

gaoyunhaii · web-flow · commit f0c1c7a9190b · 2022-06-10T11:37:14.000+08:00
This closes apache#107.
diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/HeadOperatorCheckpointAligner.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/HeadOperatorCheckpointAligner.java
@@ -23,6 +23,7 @@
 import org.apache.flink.util.function.RunnableWithException;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -109,10 +110,13 @@ List<GloballyAlignedEvent> onStateSnapshot(long checkpointId) {
     }
 
     List<GloballyAlignedEvent> onCheckpointAborted(long checkpointId) {
-        // Here we need to abort all the checkpoints <= notified checkpoint id.
-        checkState(checkpointId > latestAbortedCheckpoint);
+        if (checkpointId <= latestAbortedCheckpoint) {
+            return Collections.emptyList();
+        }
+
         latestAbortedCheckpoint = checkpointId;
 
+        // Here we need to abort all the checkpoints <= notified checkpoint id.
         Map<Long, CheckpointAlignment> abortedAlignments =
                 checkpointAlignmments.headMap(latestAbortedCheckpoint, true);
         List<GloballyAlignedEvent> events = new ArrayList<>();
