[FLINK-21954][tests] Harden JobMasterTest.testRestoringFromSavepoint …

…and .testRequestNextInputSplitWithGlobalFailover

The test instability has been introduced by FLINK-21602 because the ExecutionGraph
is now created asynchronously. This commit fixes the test instabilities in JobMasterTest
by waiting for a task submission before asserting on the state.

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

@@ -673,13 +673,24 @@ public void testRestoringFromSavepoint() throws Exception {
  // restore from the savepoint
  jobMaster.start();
 
+ final OneShotLatch taskSubmitLatch = new OneShotLatch();
+
  registerSlotsAtJobMaster(
  1,
  jobMaster.getSelfGateway(JobMasterGateway.class),
  jobGraph.getJobID(),
- new TestingTaskExecutorGatewayBuilder().createTestingTaskExecutorGateway(),
+ new TestingTaskExecutorGatewayBuilder()
+ .setSubmitTaskConsumer(
+ (taskDeploymentDescriptor, jobMasterId) -> {
+ taskSubmitLatch.trigger();
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .createTestingTaskExecutorGateway(),
  new LocalUnresolvedTaskManagerLocation());
 
+ // wait until a task has submitted because this guarantees that the ExecutionGraph has
+ // been created
+ taskSubmitLatch.await();
  final CompletedCheckpoint savepointCheckpoint =
  completedCheckpointStore.getLatestCheckpoint(false);
 
@@ -1018,13 +1029,16 @@ private void waitUntilAllExecutionsAreScheduledOrDeployed(
  final Deadline deadline = Deadline.fromNow(duration);
 
  CommonTestUtils.waitUntilCondition(
- () ->
- getExecutions(jobMasterGateway).stream()
- .allMatch(
- execution ->
- execution.getState() == ExecutionState.SCHEDULED
- || execution.getState()
- == ExecutionState.DEPLOYING),
+ () -> {
+ final Collection<AccessExecution> executions = getExecutions(jobMasterGateway);
+ return !executions.isEmpty()
+ && executions.stream()
+ .allMatch(
+ execution ->
+ execution.getState() == ExecutionState.SCHEDULED
+ || execution.getState()
+ == ExecutionState.DEPLOYING);
+ },
  deadline);
  }