[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 apache#18608.

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java

@@ -1926,10 +1926,11 @@ public void testJobMasterOnlyTerminatesAfterTheSchedulerHasClosed() throws Excep
     }
 
     @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,20 +1942,23 @@ public void testJobMasterAcceptsExcessSlotsWhenJobIsRestarting() throws Exceptio
             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()
                             .setAddress("firstTaskManager")
                             .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 void testJobMasterAcceptsExcessSlotsWhenJobIsRestarting() throws Exceptio
                                     == JobStatus.RESTARTING,
                     Deadline.fromNow(TimeUtils.toDuration(testingTimeout)));
 
-            final int numberSlots = 3;
             assertThat(
                     registerSlotsAtJobMaster(
                             numberSlots,