[BP-2.3][FLINK-40009][FLINK-40010][FLINK-40067][FLINK-40076][tests] Stabilize RescaleTimelineITCase

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,15 +269,24 @@ 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);
 
         miniCluster.submitJob(jobGraph).join();
 
         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<Rescale> rescaleHistory = getRescaleHistory(miniCluster, jobGraph);
                     return hasRescaleHistoryMetCondition(
                             rescaleHistory, 2, TerminatedReason.JOB_FINISHED);
                 },
-                10000);
+                60000);
     }
 
     @TestTemplate
@@ -351,14 +361,23 @@ 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();
         waitForVertexParallelismReachedAndJobRunning(jobGraph, JOB_VERTEX_ID, PARALLELISM);
 
         updateJobResourceRequirements(miniCluster, jobGraph, 1, PARALLELISM * 2);
 
-        assumeThat(enabledRescaleHistory(configuration)).isTrue();
         waitUntilConditionWithTimeout(
                 () -> {
                     List<Rescale> 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<Boolean, Exception> 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 {