This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 57de4676f3b [FLINK-40482][runtime] Abort checkpoint unless all tasks
are RUNNING or FINISHED
57de4676f3b is described below
commit 57de4676f3b6c6e67675ef3c4cbbacee445a1267
Author: Rui Fan <[email protected]>
AuthorDate: Thu Aug 27 15:27:26 2026 +0200
[FLINK-40482][runtime] Abort checkpoint unless all tasks are RUNNING or
FINISHED
DefaultCheckpointPlanCalculator used Execution.isFinished(), which returns
state.isTerminal(), to decide whether a task is finished. This treats any
terminal task -- FINISHED, FAILED or CANCELED -- as finished. During a
failover, FAILED/CANCELED tasks were therefore excluded from tasksToWaitFor,
producing a checkpoint plan that expects fewer acks than it should.
- calculateCheckpointPlan: replace checkAllTasksInitiated() with
checkAllTasksRunningOrFinished(), which requires every task to have an
attached Execution in RUNNING or FINISHED state, the only states a
checkpoint can be based on. Any other state aborts the checkpoint with a
graceful CheckpointException instead of building a plan from a transient
state.
- collectTaskRunningStatus: classify explicitly; a non-terminal task is
running, and a terminal task must be genuinely FINISHED.
- Remove the misleading Execution.isFinished() (its only caller).
The check is condition-based rather than mode-based, so it is correct for
both
streaming and batch. calculateCheckpointPlan and all task state transitions
run
on the JobManager main thread, so the pre-check and plan computation are
atomic.
---
.../DefaultCheckpointPlanCalculator.java | 37 ++++++++++++++++++----
.../flink/runtime/executiongraph/Execution.java | 4 ---
.../DefaultCheckpointPlanCalculatorTest.java | 33 +++++++++++++++++++
3 files changed, 63 insertions(+), 11 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculator.java
index a6874f17062..99bc5603cad 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculator.java
@@ -42,6 +42,7 @@ import java.util.concurrent.CompletionException;
import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
/**
* Default implementation for {@link CheckpointPlanCalculator}. If all tasks
are running, it
@@ -95,7 +96,7 @@ public class DefaultCheckpointPlanCalculator implements
CheckpointPlanCalculator
CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
}
- checkAllTasksInitiated();
+ checkAllTasksRunningOrFinished();
CheckpointPlan result =
context.hasFinishedTasks()
@@ -113,20 +114,32 @@ public class DefaultCheckpointPlanCalculator implements
CheckpointPlanCalculator
}
/**
- * Checks if all tasks are attached with the current Execution already.
This method should be
- * called from JobMaster main thread executor.
+ * Checks that every task is attached with the current Execution and that
this Execution is
+ * either RUNNING or FINISHED, the only states a checkpoint can be based
on. This method should
+ * be called from JobMaster main thread executor.
*
- * @throws CheckpointException if some tasks do not have attached
Execution.
+ * @throws CheckpointException if some task has no attached Execution or
is neither RUNNING nor
+ * FINISHED.
*/
- private void checkAllTasksInitiated() throws CheckpointException {
+ private void checkAllTasksRunningOrFinished() throws CheckpointException {
for (ExecutionVertex task : allTasks) {
- if (task.getCurrentExecutionAttempt() == null) {
+ Execution attempt = task.getCurrentExecutionAttempt();
+ if (attempt == null) {
throw new CheckpointException(
String.format(
"task %s of job %s is not being executed at
the moment. Aborting checkpoint.",
task.getTaskNameWithSubtaskIndex(), jobId),
CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
}
+
+ ExecutionState state = attempt.getState();
+ if (state != ExecutionState.RUNNING && state !=
ExecutionState.FINISHED) {
+ throw new CheckpointException(
+ String.format(
+ "task %s of job %s is in %s state instead of
RUNNING or FINISHED. Aborting checkpoint.",
+ task.getTaskNameWithSubtaskIndex(), jobId,
state),
+
CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
+ }
}
}
@@ -318,8 +331,18 @@ public class DefaultCheckpointPlanCalculator implements
CheckpointPlanCalculator
BitSet runningTasks = new BitSet(vertex.getTaskVertices().length);
for (int i = 0; i < vertex.getTaskVertices().length; ++i) {
- if
(!vertex.getTaskVertices()[i].getCurrentExecutionAttempt().isFinished()) {
+ Execution attempt =
vertex.getTaskVertices()[i].getCurrentExecutionAttempt();
+ ExecutionState state = attempt.getState();
+ if (state == ExecutionState.RUNNING) {
runningTasks.set(i);
+ } else {
+ // A non-RUNNING task must be FINISHED; every other state
is already rejected
+ // by checkAllTasksRunningOrFinished().
+ checkState(
+ state == ExecutionState.FINISHED,
+ "Task %s is %s, not FINISHED",
+ attempt.getAttemptId(),
+ state);
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 430cd9ee76d..46d5a8d60b0 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -376,10 +376,6 @@ public class Execution
return this.stateEndTimestamps[state.ordinal()];
}
- public boolean isFinished() {
- return state.isTerminal();
- }
-
@Nullable
public JobManagerTaskRestore getTaskRestore() {
return taskRestore;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculatorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculatorTest.java
index faa42c20b32..cbe68312ecc 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculatorTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/DefaultCheckpointPlanCalculatorTest.java
@@ -37,6 +37,8 @@ import
org.apache.flink.testutils.executor.TestExecutorExtension;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import java.util.ArrayList;
import java.util.Arrays;
@@ -166,6 +168,37 @@ class DefaultCheckpointPlanCalculatorTest {
// then: The plan failed because one task didn't have RUNNING state.
}
+ @ParameterizedTest
+ @EnumSource(
+ value = ExecutionState.class,
+ names = {"FAILED", "CANCELED", "CANCELING"})
+ void
testPlanAbortedWhenTaskIsFailingOverWhileOthersFinished(ExecutionState
failoverState)
+ throws Exception {
+ ExecutionGraph graph =
+ createExecutionGraph(
+ Arrays.asList(
+ new VertexDeclaration(1, of(0)), new
VertexDeclaration(1, of())),
+ Collections.emptyList());
+
+ chooseJobVertex(graph, 1)
+ .getTaskVertices()[0]
+ .getCurrentExecutionAttempt()
+ .transitionState(failoverState);
+
+ DefaultCheckpointPlanCalculator planCalculator =
createCheckpointPlanCalculator(graph);
+
+ assertThatFuture(planCalculator.calculateCheckpointPlan())
+ .eventuallyFailsWith(ExecutionException.class)
+ .havingCause()
+ .isInstanceOfSatisfying(
+ CheckpointException.class,
+ e ->
+ assertThat(e.getCheckpointFailureReason())
+ .isEqualTo(
+ CheckpointFailureReason
+
.NOT_ALL_REQUIRED_TASKS_RUNNING));
+ }
+
private void runWithNotRunningTask(
boolean isRunningVertexSource, boolean isNotRunningVertexSource)
throws Exception {
for (ExecutionState notRunningState :
complementOf(EnumSet.of(ExecutionState.RUNNING))) {
