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..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 @@ -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; @@ -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 @@ -351,6 +361,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 +378,6 @@ void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws Exception updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2); - assumeThat(enabledRescaleHistory(configuration)).isTrue(); waitUntilConditionWithTimeout( () -> { List rescaleHistory = getRescaleHistory(miniCluster, jobGraph); @@ -367,7 +386,7 @@ void testRescaleTerminatedByNoResourcesOrNoParallelismsChange() throws Exception 2, TerminatedReason.NO_RESOURCES_OR_PARALLELISMS_CHANGE); }, - 20000); + 60000); } @TestTemplate @@ -396,44 +415,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); @@ -590,7 +577,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(); @@ -700,15 +691,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( @@ -734,6 +731,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 {