From 6ec03c995a625cc22c3cdeb5076b0ca2d5329c3e Mon Sep 17 00:00:00 2001
From: Efrat Levitan <elevitan@confluent.io>
Date: Wed, 22 Jul 2026 13:32:09 +0300
Subject: [PATCH] [FLINK-33186][checkpoint/tests] Fix flakiness in
 CheckpointAfterAllTasksFinishedITCase.testRestoreAfterSomeTasksFinished

I traced the flakiness to task finished after the checkpoint plan was calculated
but before triggerCheckpoint. The task is already removed from TaskExecutor#taskSlotTable
So TaskExecutor#triggerCheckpoint fails the checkpoint with error
'TaskManager received a checkpoint request for unknown task <>'.

Production systems are expected to tolerate an occasional checkpoint failure
So we can address the flakiness without changes this known limitation for
tasks finished mid checkpoint.

the test provisions 2 sources passA that finishes quickly and passB that blocks forever
and non-deterministically waits for at least one of passA subtasks to finish, which introduces
the race.
The fix here makes CheckpointAfterAllTasksFinishedITCase.testRestoreAfterSomeTasksFinished
wait for all subtasks of passA to finish before triggering savepoint, to avoid the race
but still achieve the test goal with passB subtasks still locking.
---
 .../checkpointing/CheckpointAfterAllTasksFinishedITCase.java    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
index 6ff385a8ba45b..badf21f736dd4 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/CheckpointAfterAllTasksFinishedITCase.java
@@ -112,7 +112,7 @@ void testRestoreAfterSomeTasksFinished() throws Exception {
                     miniCluster,
                     jobGraph.getJobID(),
                     findVertexByName(jobGraph, "passA -> Sink: sinkA").getID(),
-                    false);
+                    true);
 
             String savepointPath =
                     miniCluster