This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new be19c67322a [FLINK-38534][runtime/tests] Fix LocalRecoveryTest racing
the async task deployment
be19c67322a is described below
commit be19c67322a77995a01dc800f2921ea745b3ccc4
Author: Martijn Visser <[email protected]>
AuthorDate: Fri Jun 12 12:15:47 2026 +0200
[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)
---
.../runtime/scheduler/SchedulerTestingUtils.java | 32 ++++++++++++++++++++++
.../scheduler/adaptive/LocalRecoveryTest.java | 16 +++++++----
2 files changed, 43 insertions(+), 5 deletions(-)
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 49a81d6fd2d..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
@@ -362,6 +362,38 @@ public class SchedulerTestingUtils {
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 bdaf69c3f31..4227afa3bf3 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,7 @@ 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;
@@ -103,11 +104,6 @@ public class LocalRecoveryTest extends
AdaptiveSchedulerTestBase {
// Transition job and all subtasks to RUNNING state.
waitForJobStatusRunning(scheduler);
- 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.
final ExecutionGraph executionGraph =
scheduler
.getState()
@@ -115,6 +111,16 @@ public class LocalRecoveryTest extends
AdaptiveSchedulerTestBase {
.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
