This is an automated email from the ASF dual-hosted git repository. MartijnVisser pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit ad2dee480a697d831d79ca1fd6cc501b198c7765 Author: Martijn Visser <[email protected]> AuthorDate: Mon Jun 29 10:59:48 2026 +0200 [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) --- .../adaptive/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 cbf32d7467e..a25d0c97223 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 @@ class RescaleTimelineITCase { @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 @@ class RescaleTimelineITCase { updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2); - assumeThat(enabledRescaleHistory(configuration)).isTrue(); waitUntilConditionWithTimeout( () -> { List<Rescale> rescaleHistory = getRescaleHistory(miniCluster, jobGraph); @@ -367,7 +376,7 @@ class RescaleTimelineITCase { 2, TerminatedReason.NO_RESOURCES_OR_PARALLELISMS_CHANGE); }, - 20000); + 60000); } @TestTemplate @@ -396,44 +405,12 @@ class RescaleTimelineITCase { // 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 @@ class RescaleTimelineITCase { == 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 {
