[FLINK-34274][runtime] Implicitly disable resource wait timeout for A…

…daptiveSchedulerTest (apache#24238)
aiven · Feb 28, 2024 · bfaa75a · bfaa75a
1 parent 94b5f03
commit bfaa75a
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 9 deletions.
diff --git a/...e/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java b/...e/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java
@@ -52,6 +52,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Function;
 
 /** Builder for {@link AdaptiveScheduler}. */
 public class AdaptiveSchedulerBuilder {
@@ -117,6 +118,12 @@ public AdaptiveSchedulerBuilder setJobMasterConfiguration(
         return this;
     }
 
+    public AdaptiveSchedulerBuilder withConfigurationOverride(
+            Function<Configuration, Configuration> modifyFn) {
+        this.jobMasterConfiguration = modifyFn.apply(jobMasterConfiguration);
+        return this;
+    }
+
     public AdaptiveSchedulerBuilder setUserCodeLoader(final ClassLoader userCodeLoader) {
         this.userCodeLoader = userCodeLoader;
         return this;

diff --git a/...time/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/...time/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -1221,6 +1221,13 @@ void testInitialRequirementLowerBoundBeyondAvailableSlotsCausesImmediateFailure(
 
         final AdaptiveScheduler scheduler =
                 prepareSchedulerWithNoTimeouts(jobGraph, declarativeSlotPool)
+                        .withConfigurationOverride(
+                                conf -> {
+                                    conf.set(
+                                            JobManagerOptions.RESOURCE_WAIT_TIMEOUT,
+                                            Duration.ofMillis(1));
+                                    return conf;
+                                })
                         .setJobResourceRequirements(initialJobResourceRequirements)
                         .build();
 
@@ -1261,14 +1268,6 @@ void testRequirementLowerBoundDecreaseAfterResourceScarcityBelowAvailableSlots()
         startJobWithSlotsMatchingParallelism(
                 scheduler, declarativeSlotPool, taskManagerGateway, availableSlots);
 
-        // at this point we'd ideally check that the job is stuck in WaitingForResources, but we
-        // can't differentiate between waiting due to the minimum requirements not being fulfilled
-        // and the resource timeout not being elapsed
-        // We just continue here, as the following tests validate that the lower bound can prevent
-        // a job from running:
-        // - #testInitialRequirementLowerBoundBeyondAvailableSlotsCausesImmediateFailure()
-        // - #testRequirementLowerBoundIncreaseBeyondCurrentParallelismAttemptsImmediateRescale()
-
         // unlock job by decreasing the parallelism
         JobResourceRequirements newJobResourceRequirements =
                 createRequirementsWithLowerAndUpperParallelism(availableSlots, PARALLELISM);
@@ -1280,7 +1279,8 @@ void testRequirementLowerBoundDecreaseAfterResourceScarcityBelowAvailableSlots()
 
     private static Configuration createConfigurationWithNoTimeouts() {
         return new Configuration()
-                .set(JobManagerOptions.RESOURCE_WAIT_TIMEOUT, Duration.ofMillis(1L))
+                .set(JobManagerOptions.RESOURCE_WAIT_TIMEOUT, Duration.ofMillis(-1L))
+                .set(JobManagerOptions.RESOURCE_STABILIZATION_TIMEOUT, Duration.ofMillis(1L))
                 .set(JobManagerOptions.SCHEDULER_SCALING_INTERVAL_MIN, Duration.ofMillis(1L));
     }