This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.2 by this push:
new 459a16c6066 [FLINK-38534][runtime/tests] Fix flaky LocalRecoveryTest
by waiting for tasks to reach RUNNING state (#29751)
459a16c6066 is described below
commit 459a16c6066d1fe8d7ce0c779e44185292763e0e
Author: Rui Fan <[email protected]>
AuthorDate: Tue Nov 3 20:37:21 2026 +0100
[FLINK-38534][runtime/tests] Fix flaky LocalRecoveryTest by waiting for
tasks to reach RUNNING state (#29751)
* [FLINK-38534][runtime/tests] Fix flaky LocalRecoveryTest by waiting for
tasks to reach RUNNING state (#27767)
(cherry picked from commit df4f8c48d6042f4e1a2ccf5282ba0ad8cf541fbd)
* [FLINK-38534][runtime/tests] Fix LocalRecoveryTest racing the async task
deployment
The test forces executions to RUNNING while the scheduler is still
creating task deployment descriptors; deployment is then rejected, the
job restarts, and the manual checkpoint never registers. Wait for
descriptor creation before forcing RUNNING, keyed off the descriptor
future because executions are observable in DEPLOYING before the future
is assigned. The earlier FLINK-38534 fix waited for RUNNING after the
forced transition and misses this race.
Generated-by: Claude Opus 4.8 (1M context)
(cherry picked from commit e009616411c127c36d95ba38a7edb66c22eda62f)
---------
Co-authored-by: mukul-8 <[email protected]>
Co-authored-by: Martijn Visser
<[email protected]>
---
.../runtime/scheduler/SchedulerTestingUtils.java | 54 ++++++++++++++++++++++
.../scheduler/adaptive/LocalRecoveryTest.java | 19 ++++++++
2 files changed, 73 insertions(+)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
index 0b53800e159..1257f5a1285 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SchedulerTestingUtils.java
@@ -340,6 +340,60 @@ public class SchedulerTestingUtils {
RETRY_ATTEMPTS);
}
+ /**
+ * Waits until all task executions in the given ExecutionGraph reach
RUNNING state.
+ *
+ * @param executionGraph the ExecutionGraph to check
+ * @throws Exception if the condition is not met within the timeout period
+ */
+ public static void waitForAllTasksRunning(final ExecutionGraph
executionGraph)
+ throws Exception {
+ waitUntilCondition(
+ () -> {
+ for (ExecutionVertex vertex :
executionGraph.getAllExecutionVertices()) {
+ if (vertex.getCurrentExecutionAttempt().getState()
+ != ExecutionState.RUNNING) {
+ return false;
+ }
+ }
+ return true;
+ },
+ RETRY_INTERVAL_MILLIS,
+ RETRY_ATTEMPTS);
+ }
+
+ /**
+ * Waits until the task deployment descriptor of every current execution
has been created. Call
+ * this before forcing executions into a later state; deployment is
rejected once an execution
+ * leaves DEPLOYING.
+ *
+ * <p>Unlike {@code
ExecutionUtils#waitForTaskDeploymentDescriptorsCreation}, this may run
+ * concurrently with {@code Execution.deploy()}: executions are observable
in DEPLOYING before
+ * the descriptor future is assigned, so the {@code IllegalStateException}
thrown while the
+ * future is unset is treated as "not yet ready".
+ *
+ * @param executionGraph the ExecutionGraph to wait for
+ * @throws Exception if the condition is not met within the timeout period
+ */
+ public static void waitForAllTasksDeploymentDescriptorsCreated(
+ final ExecutionGraph executionGraph) throws Exception {
+ for (ExecutionVertex vertex :
executionGraph.getAllExecutionVertices()) {
+ waitUntilCondition(
+ () -> {
+ try {
+ vertex.getCurrentExecutionAttempt()
+ .getTddCreationDuringDeployFuture()
+ .join();
+ return true;
+ } catch (IllegalStateException
deploymentNotStartedYet) {
+ return false;
+ }
+ },
+ RETRY_INTERVAL_MILLIS,
+ RETRY_ATTEMPTS);
+ }
+ }
+
private static ExecutionJobVertex getJobVertex(
DefaultScheduler scheduler, JobVertexID jobVertexId) {
final ExecutionVertexID id = new ExecutionVertexID(jobVertexId, 0);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/LocalRecoveryTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/LocalRecoveryTest.java
index 3f08094fe50..934dacaf27c 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/LocalRecoveryTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/LocalRecoveryTest.java
@@ -56,6 +56,8 @@ import java.util.stream.Collectors;
import static
org.apache.flink.runtime.checkpoint.CheckpointCoordinatorTestingUtils.generateKeyGroupState;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.acknowledgePendingCheckpoint;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.setAllExecutionsToRunning;
+import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.waitForAllTasksDeploymentDescriptorsCreated;
+import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.waitForAllTasksRunning;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.waitForCheckpointInProgress;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.waitForCompletedCheckpoint;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.waitForJobStatusRunning;
@@ -99,7 +101,24 @@ public class LocalRecoveryTest extends
AdaptiveSchedulerTestBase {
// Transition job and all subtasks to RUNNING state.
waitForJobStatusRunning(scheduler);
+ final ExecutionGraph executionGraph =
+ scheduler
+ .getState()
+ .as(StateWithExecutionGraph.class)
+ .map(StateWithExecutionGraph::getExecutionGraph)
+ .orElseThrow(
+ () -> new
IllegalStateException("ExecutionGraph not available"));
+
+ // Forcing RUNNING while the asynchronous descriptor creation is still
in flight fails the
+ // deployment and restarts the job, so the checkpoint below would
never register.
+ waitForAllTasksDeploymentDescriptorsCreated(executionGraph);
+
runInMainThread(() -> setAllExecutionsToRunning(scheduler));
+ // Wait for all task executions to actually reach RUNNING state before
triggering
+ // checkpoint.
+ // In slower CI environments, state transitions may not complete
immediately, causing
+ // checkpoint triggers to be rejected if tasks are still in
DEPLOYING/INITIALIZING state.
+ waitForAllTasksRunning(executionGraph);
// Trigger a checkpoint
CompletableFuture<CompletedCheckpoint> completedCheckpointFuture =
