From 9971a000087d8b79d202fe647931accaef296c57 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:56:37 +0200 Subject: [PATCH 1/4] [FLINK-40009][tests] Enforce timeout in RescaleTimelineITCase wait helper waitUntilConditionWithTimeout wrapped an unbounded CommonTestUtils#waitUntilCondition in CompletableFuture#runAsync and bounded it with get(timeout). On JDK 11, when the test runs on a ForkJoinPool worker (JUnit 5's test executor), CompletableFuture#timedGet help-executes the async task inline via ForkJoinPool#helpAsyncBlocker, so the never-ending poll loop runs on the waiting thread and the timeout never fires. A CI thread dump of the JDK 11 cron leg showed the call still parked after 978s on a 20s budget, hanging the fork until the watchdog killed it. JDK 17 and 21 changed helpAsyncBlocker and do not hang. Poll synchronously on the calling thread via CommonTestUtils#waitUtil so the timeout is always enforced and no task is leaked into the common pool. Generated-by: Claude Opus 4.8 (1M context) (cherry picked from commit 1268adfdb03ff9dca2ce53a224974db6918b66d5) --- .../timeline/RescaleTimelineITCase.java | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java index 11711208453c17..cbf32d7467e955 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java @@ -57,9 +57,7 @@ import java.util.Collection; import java.util.List; import java.util.Map; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -77,6 +75,8 @@ class RescaleTimelineITCase { private static final int NUMBER_TASK_MANAGERS = 2; private static final int PARALLELISM = NUMBER_SLOTS_PER_TASK_MANAGER * NUMBER_TASK_MANAGERS; private static final JobVertexID JOB_VERTEX_ID = new JobVertexID(); + // Matches the default poll interval of the CommonTestUtils#waitUntilCondition it replaced. + private static final long RETRY_INTERVAL_MILLIS = 100L; @Parameter private Configuration configuration; private MiniClusterResource miniClusterResource; @@ -700,15 +700,21 @@ private void assertVertexParallelismRescaleNotNullBesidesPreRelatedFields( private void waitUntilConditionWithTimeout( SupplierWithException condition, long timeoutMillis) throws Exception { - CompletableFuture.runAsync( - () -> { - try { - CommonTestUtils.waitUntilCondition(condition); - } catch (Exception e) { - throw new RuntimeException(e); - } - }) - .get(timeoutMillis, TimeUnit.MILLISECONDS); + // Poll synchronously via waitUtil. The previous CompletableFuture#runAsync + get(timeout) + // wrapper hangs on JDK 11: on a ForkJoinPool worker (JUnit's executor) timedGet + // help-executes the unbounded poll loop inline on the waiting thread, so the timeout + // never fires. + org.apache.flink.core.testutils.CommonTestUtils.waitUtil( + () -> { + try { + return condition.get(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }, + Duration.ofMillis(timeoutMillis), + Duration.ofMillis(RETRY_INTERVAL_MILLIS), + "Condition was not met within " + timeoutMillis + " ms."); } private void waitForVertexParallelismReachedAndJobRunning( From d8a525c9e6eb11389e8615af8df8f210836d945e Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Mon, 29 Jun 2026 10:59:48 +0200 Subject: [PATCH 2/4] [FLINK-40010][tests] Fix race in RescaleTimelineITCase.testRescaleTerminatedByNoResourcesOrNoParallelismsChange NO_RESOURCES_OR_PARALLELISMS_CHANGE is stamped by DefaultStateTransitionManager only on the rescale tracked when the manager (re-)enters its Idling phase. With the short shared cooldown the cooldown can elapse and reach Idling before the requirements-update RPC is processed, so the UPDATE_REQUIREMENT rescale is created after Idling was entered and never receives the terminal reason; it stays in-progress until teardown cancels it (JOB_CANCELED) and the wait times out. Rebuild the fixture cluster with a cooldown that comfortably outlasts the synchronous update RPC so the update is processed in Cooldown and routed back through Idling, where the reason is recorded. Unlike testRescaleTerminatedByResourceRequirementsUpdated this case must wait out the whole cooldown before the reason appears, so it uses a 10s cooldown (not 60s) and a 60s wait budget for a >5x margin; the resource-stabilization timeout stays at the short fixture default. Extract the shared cluster-rebuild block into rebuildClusterWithExecutingTimeouts and reuse it from both tests. Generated-by: Claude Opus 4.8 (1M context) (cherry picked from commit ad2dee480a697d831d79ca1fd6cc501b198c7765) --- .../timeline/RescaleTimelineITCase.java | 81 ++++++++++--------- 1 file changed, 41 insertions(+), 40 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java index cbf32d7467e955..a25d0c9722368f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java @@ -351,6 +351,16 @@ void testRescaleTerminatedByJobFailed() throws Exception { @TestTemplate void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws Exception { + // This case only asserts on the recorded rescale history; skip the disabled-history + // parameter before the cluster rebuild below so it does not pay for an unused cluster. + assumeThat(enabledRescaleHistory(configuration)).isTrue(); + + // NO_RESOURCES_OR_PARALLELISMS_CHANGE is only stamped when the update RPC is processed + // during Cooldown and the manager re-enters Idling. Unlike the sibling + // testRescaleTerminatedByResourceRequirementsUpdated, this case must wait out the whole + // cooldown, so it cannot reuse 60s: 10s outlasts the RPC yet stays within the 60s budget. + rebuildClusterWithExecutingTimeouts(Duration.ofSeconds(10), Duration.ofMillis(50)); + final MiniCluster miniCluster = miniClusterResource.getMiniCluster(); final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM); miniCluster.submitJob(jobGraph).join(); @@ -358,7 +368,6 @@ void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws Exception updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2); - assumeThat(enabledRescaleHistory(configuration)).isTrue(); waitUntilConditionWithTimeout( () -> { List rescaleHistory = getRescaleHistory(miniCluster, jobGraph); @@ -367,7 +376,7 @@ void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws Exception 2, TerminatedReason.NO_RESOURCES_OR_PARALLELISMS_CHANGE); }, - 20000); + 60000); } @TestTemplate @@ -396,44 +405,12 @@ void testRescaleTerminatedByResourceRequirementsUpdated() throws Exception { // parameter before the cluster rebuild below so it does not pay for an unused cluster. assumeThat(enabledRescaleHistory(configuration)).isTrue(); - // This test asserts that the second resource-requirements update terminates the in-progress - // rescale started by the first update with terminal reason RESOURCE_REQUIREMENTS_UPDATED. - // For that to happen the second update must be processed while the first rescale is still - // in-progress: AdaptiveScheduler#recordRescaleForNewResourceRequirements only sets the - // RESOURCE_REQUIREMENTS_UPDATED reason via RescaleTimeline#updateRescale, which is a no-op - // once the current rescale is already terminated (DefaultRescaleTimeline#isIdling). - // - // The upper bound exceeds the available slots, so the first rescale cannot change the - // parallelism. With the short cooldown/stabilization timeouts shared by the other cases the - // DefaultStateTransitionManager re-enters its Idling phase and the Idling constructor - // terminates that rescale with NO_RESOURCES_OR_PARALLELISMS_CHANGE (a legitimate terminal - // reason for an in-progress rescale that cannot change parallelism). That termination is - // driven by wall-clock timers that start the moment the first rescale is recorded, so no - // amount of waiting between the two updates can guarantee the second update wins the race; - // waiting only consumes the same budget. Widening cooldown and stabilization for this case - // is therefore the only deterministic test-side fix: it keeps the in-progress rescale alive - // far longer than the single synchronous RPC round trip between the two updates. - // - // Rebuild the shared fixture cluster in place rather than starting a second one on top of - // it, so only one cluster is ever running and the @AfterEach teardown still applies. 60s is - // an intentionally generous bound (the suite already uses second-scale guards for slow CI); - // the test completes as soon as the second update lands, so a large value has no cost. - miniClusterResource.after(); - final Configuration testConfiguration = new Configuration(configuration); - testConfiguration.set( - JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING, - Duration.ofSeconds(60)); - testConfiguration.set( - JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT, - Duration.ofSeconds(60)); - miniClusterResource = - new MiniClusterResource( - new MiniClusterResourceConfiguration.Builder() - .setConfiguration(testConfiguration) - .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER) - .setNumberTaskManagers(NUMBER_TASK_MANAGERS) - .build()); - miniClusterResource.before(); + // The second update only terminates the first rescale with RESOURCE_REQUIREMENTS_UPDATED + // while that rescale is still in-progress; once it is terminated, updateRescale is a no-op. + // With the short shared cooldown the manager re-enters Idling and terminates it on a + // wall-clock timer first, so waiting cannot win the race. Widening cooldown and + // stabilization to 60s keeps the rescale in-progress far longer than the RPC round trip. + rebuildClusterWithExecutingTimeouts(Duration.ofSeconds(60), Duration.ofSeconds(60)); final MiniCluster miniCluster = miniClusterResource.getMiniCluster(); final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM); @@ -740,6 +717,30 @@ private void waitForVertexParallelismReachedAndJobRunning( == JobStatus.RUNNING); } + /** + * Rebuilds the shared fixture cluster in place with the given executing-phase cooldown and + * resource-stabilization timeouts. Rebuilding in place rather than starting a second cluster + * keeps a single cluster running so the {@link AfterEach} teardown still applies. + */ + private void rebuildClusterWithExecutingTimeouts( + Duration cooldown, Duration resourceStabilizationTimeout) throws Exception { + miniClusterResource.after(); + final Configuration testConfiguration = new Configuration(configuration); + testConfiguration.set( + JobManagerOptions.SCHEDULER_EXECUTING_COOLDOWN_AFTER_RESCALING, cooldown); + testConfiguration.set( + JobManagerOptions.SCHEDULER_EXECUTING_RESOURCE_STABILIZATION_TIMEOUT, + resourceStabilizationTimeout); + miniClusterResource = + new MiniClusterResource( + new MiniClusterResourceConfiguration.Builder() + .setConfiguration(testConfiguration) + .setNumberSlotsPerTaskManager(NUMBER_SLOTS_PER_TASK_MANAGER) + .setNumberTaskManagers(NUMBER_TASK_MANAGERS) + .build()); + miniClusterResource.before(); + } + private void updateJobResourceRequirements( MiniCluster miniCluster, JobGraph jobGraph, int lowerBound, int upperBound) throws ExecutionException, InterruptedException { From f6f400bdc2b3820d350edcedb4cdecf6974e07c0 Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Sun, 5 Jul 2026 19:59:52 +0200 Subject: [PATCH 3/4] [FLINK-40076][tests] Fix race in RescaleTimelineITCase.testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause After terminating a TaskManager to trigger a recoverable failover, the test waited via waitForVertexParallelismReachedAndJobRunning for PARALLELISM. That condition is already satisfied by the stale pre-failover state, so the wait returns immediately without synchronizing with the failover. The snapshot could therefore be taken before AdaptiveScheduler merged the failover into the in-progress UPDATE_REQUIREMENT rescale and re-stamped its trigger cause, letting the test observe UPDATE_REQUIREMENT instead of the expected RECOVERABLE_FAILOVER. Wait for the job to be RUNNING again at the parallelism the single remaining TaskManager can host (NUMBER_SLOTS_PER_TASK_MANAGER) instead. The merge re-stamps the trigger cause in goToRestarting, before the job runs again at that reduced parallelism, so the job coming back up synchronizes the snapshot with the merge. The assertion is unchanged. Generated-by: Claude Opus 4.8 (via Claude Code) (cherry picked from commit 7f770b87178b66781708b92beb3540e5257c7c9f) --- .../scheduler/adaptive/timeline/RescaleTimelineITCase.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java index a25d0c9722368f..96328abf426818 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java @@ -567,7 +567,11 @@ void testRecordNonTerminatedRescaleMergingWithNewRecoverableFailureTriggerCause( miniCluster.terminateTaskManager(0); - waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID, PARALLELISM); + // Wait for the failover to complete before snapshotting: the merge that re-stamps the + // trigger cause to RECOVERABLE_FAILOVER runs before the job is RUNNING again at the + // reduced parallelism (one TaskManager left). + waitForVertexParallelismReachedAndJobRunning( + jobGraph, JOB_VERTEX_ID, NUMBER_SLOTS_PER_TASK_MANAGER); final ExecutionGraphInfo executionGraphInfo = miniCluster.getExecutionGraphInfo(jobGraph.getJobID()).join(); From 12cdba7ca549c92f712fc525a28ee97002f5206d Mon Sep 17 00:00:00 2001 From: Martijn Visser <2989614+MartijnVisser@users.noreply.github.com> Date: Fri, 3 Jul 2026 21:20:20 +0200 Subject: [PATCH 4/4] [FLINK-40067][tests] Fix race in RescaleTimelineITCase.testRescaleTerminatedByJobFinished On a loaded machine the cooldown-driven Idling transition terminates the in-progress rescale with NO_RESOURCES_OR_PARALLELISMS_CHANGE before the job finishes, so goToFinished's later JOB_FINISHED stamp is dropped and the wait can never succeed. Widen the fixture cooldown to keep the rescale in-progress until the job finishes, mirroring FLINK-39903/FLINK-40010. Generated-by: Claude Opus 4.8 (1M context) (cherry picked from commit edba4901475bfd0696d4e95f1c9ca049924b7a81) --- .../adaptive/timeline/RescaleTimelineITCase.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java index 96328abf426818..e5920f282cbcff 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/timeline/RescaleTimelineITCase.java @@ -269,6 +269,17 @@ void testRescaleTerminatedBySucceeded() {} @TestTemplate void testRescaleTerminatedByJobFinished() throws Exception { + // This case only asserts on the recorded rescale history; skip the disabled-history + // parameter before the cluster rebuild below so it does not pay for an unused cluster. + assumeThat(enabledRescaleHistory(configuration)).isTrue(); + + // With the short shared cooldown the DefaultStateTransitionManager re-enters Idling on a + // wall-clock timer and terminates the in-progress rescale with + // NO_RESOURCES_OR_PARALLELISMS_CHANGE before the job finishes; goToFinished then finds it + // already terminated and its JOB_FINISHED stamp is ignored, so waiting cannot win. + // Widen the cooldown to keep the rescale in-progress until the job finishes. + rebuildClusterWithExecutingTimeouts(Duration.ofSeconds(60), Duration.ofSeconds(60)); + final MiniCluster miniCluster = miniClusterResource.getMiniCluster(); final JobGraph jobGraph = createBlockingJobGraph(PARALLELISM); @@ -276,8 +287,6 @@ void testRescaleTerminatedByJobFinished() throws Exception { waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID, PARALLELISM); - assumeThat(enabledRescaleHistory(configuration)).isTrue(); - updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2); // The upper bound (PARALLELISM * 2) exceeds the available slots, so this rescale is only @@ -289,13 +298,14 @@ void testRescaleTerminatedByJobFinished() throws Exception { OnceBlockingNoOpInvokable.unblock(); + // Generous budget: on a loaded CI leg the unblock-to-finish window can itself exceed 10s. waitUntilConditionWithTimeout( () -> { List rescaleHistory = getRescaleHistory(miniCluster, jobGraph); return hasRescaleHistoryMetCondition( rescaleHistory, 2, TerminatedReason.JOB_FINISHED); }, - 10000); + 60000); } @TestTemplate