[flink] branch master updated: [FLINK-25925][tests] Require slot pool to only accept the same set of slots as before

[trohrmann]

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

trohrmann 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 2317ab4  [FLINK-25925][tests] Require slot pool to only accept the 
same set of slots as before
2317ab4 is described below

commit 2317ab4f29180951712d3606ec7da156dc00d3ea
Author: Till Rohrmann <trohrm...@apache.org>
AuthorDate: Wed Feb 2 14:43:27 2022 +0100

    [FLINK-25925][tests] Require slot pool to only accept the same set of slots 
as before
    
    Instead of also accepting excess slots, we only need to require the same 
set of slots
    as before. This commit changes the 
JobMasterTest.testJobMasterAcceptsExcessSlotsWhenJobIsRestarting
    accordingly.
    
    This closes #18608.
---
 .../org/apache/flink/runtime/jobmaster/JobMasterTest.java | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 1a44fa0..50bac5c 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -1926,10 +1926,11 @@ public class JobMasterTest extends TestLogger {
     }
 
     @Test
-    public void testJobMasterAcceptsExcessSlotsWhenJobIsRestarting() throws 
Exception {
+    public void testJobMasterAcceptsSlotsWhenJobIsRestarting() throws 
Exception {
         configuration.set(RestartStrategyOptions.RESTART_STRATEGY, 
"fixed-delay");
         configuration.set(
                 RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_DELAY, 
Duration.ofDays(1));
+        final int numberSlots = 1;
         final JobMaster jobMaster =
                 new JobMasterBuilder(jobGraph, rpcService)
                         .withConfiguration(configuration)
@@ -1941,13 +1942,10 @@ public class JobMasterTest extends TestLogger {
             final JobMasterGateway jobMasterGateway =
                     jobMaster.getSelfGateway(JobMasterGateway.class);
 
-            assertThat(
-                    jobMasterGateway.requestJobStatus(testingTimeout).get(), 
is(JobStatus.RUNNING));
-
             final LocalUnresolvedTaskManagerLocation 
unresolvedTaskManagerLocation =
                     new LocalUnresolvedTaskManagerLocation();
             registerSlotsAtJobMaster(
-                    1,
+                    numberSlots,
                     jobMasterGateway,
                     jobGraph.getJobID(),
                     new TestingTaskExecutorGatewayBuilder()
@@ -1955,6 +1953,12 @@ public class JobMasterTest extends TestLogger {
                             .createTestingTaskExecutorGateway(),
                     unresolvedTaskManagerLocation);
 
+            CommonTestUtils.waitUntilCondition(
+                    () ->
+                            
jobMasterGateway.requestJobStatus(testingTimeout).get()
+                                    == JobStatus.RUNNING,
+                    Deadline.fromNow(TimeUtils.toDuration(testingTimeout)));
+
             jobMasterGateway.disconnectTaskManager(
                     unresolvedTaskManagerLocation.getResourceID(),
                     new FlinkException("Test exception."));
@@ -1965,7 +1969,6 @@ public class JobMasterTest extends TestLogger {
                                     == JobStatus.RESTARTING,
                     Deadline.fromNow(TimeUtils.toDuration(testingTimeout)));
 
-            final int numberSlots = 3;
             assertThat(
                     registerSlotsAtJobMaster(
                             numberSlots,