You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2022/06/10 03:37:18 UTC
[flink-ml] branch master updated: [FLINK-26801] Support duplicate checkpoint aborted messages
This is an automated email from the ASF dual-hosted git repository.
lindong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new f0c1c7a [FLINK-26801] Support duplicate checkpoint aborted messages
f0c1c7a is described below
commit f0c1c7a9190b802eb79f2f299bbb9bd22f4eb79d
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Fri Jun 10 11:37:14 2022 +0800
[FLINK-26801] Support duplicate checkpoint aborted messages
This closes #107.
---
.../flink/iteration/operator/HeadOperatorCheckpointAligner.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
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
index c6e86db..f138992 100644
--- 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.iteration.operator.event.GloballyAlignedEvent;
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 @@ class HeadOperatorCheckpointAligner {
}
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<>();