(flink) branch master updated: [FLINK-34274][runtime] Implicitly disable resource wait timeout for AdaptiveSchedulerTest (#24238)

[mapohl]

This is an automated email from the ASF dual-hosted git repository.

mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new bfaa75a4184 [FLINK-34274][runtime] Implicitly disable resource wait 
timeout for AdaptiveSchedulerTest (#24238)
bfaa75a4184 is described below

commit bfaa75a418432162700996053319d1a5c1e72927
Author: David Moravek <d...@apache.org>
AuthorDate: Wed Feb 28 15:22:48 2024 +0100

    [FLINK-34274][runtime] Implicitly disable resource wait timeout for 
AdaptiveSchedulerTest (#24238)
---
 .../scheduler/adaptive/AdaptiveSchedulerBuilder.java   |  7 +++++++
 .../scheduler/adaptive/AdaptiveSchedulerTest.java      | 18 +++++++++---------
 2 files changed, 16 insertions(+), 9 deletions(-)

diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java
index b814144ad96..fca3c7a8548 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerBuilder.java
@@ -52,6 +52,7 @@ import javax.annotation.Nullable;
 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 class AdaptiveSchedulerBuilder {
         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/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index 7b028991f37..e2260d1c19c 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -1221,6 +1221,13 @@ public class AdaptiveSchedulerTest {
 
         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 @@ public class AdaptiveSchedulerTest {
         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 @@ public class AdaptiveSchedulerTest {
 
     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));
     }