[FLINK-38748][runtime] Enhance the switching logic of AdaptiveScheduler from Restarting to CreatingExecutionGraph and accelerate the transition.

Author: RocMarshal

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java

@@ -114,7 +114,8 @@ void onGloballyTerminalState(JobStatus globallyTerminalState) {
     }
 
     private void goToSubsequentState() {
-        if (availableParallelismNotChanged(restartWithParallelism)) {
+        if (availableParallelismNotChanged(restartWithParallelism)
+                || context.hasDesiredResources()) {
             context.goToCreatingExecutionGraph(getExecutionGraph());
         } else {
             context.goToWaitingForResources(getExecutionGraph());
@@ -163,6 +164,13 @@ interface Context
          * slots.
          */
         Optional<VertexParallelism> getAvailableVertexParallelism();
+
+        /**
+         * Checks whether we have the desired resources.
+         *
+         * @return {@code true} if we have enough resources; otherwise {@code false}
+         */
+        boolean hasDesiredResources();
     }
 
     static class Factory implements StateFactory<Restarting> {

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/MockRestartingContext.java

@@ -51,6 +51,8 @@ class MockRestartingContext extends MockStateWithExecutionGraphContext
 
     @Nullable private VertexParallelism availableVertexParallelism;
 
+    private boolean hasDesiredResources = false;
+
     public void setExpectCancelling(Consumer<ExecutingTest.CancellingArguments> asserter) {
         cancellingStateValidator.expectInput(asserter);
     }
@@ -68,6 +70,15 @@ public void setAvailableVertexParallelism(
         this.availableVertexParallelism = availableVertexParallelism;
     }
 
+    public void setHasDesiredResources(boolean hasDesiredResources) {
+        this.hasDesiredResources = hasDesiredResources;
+    }
+
+    @Override
+    public boolean hasDesiredResources() {
+        return hasDesiredResources;
+    }
+
     @Override
     public void goToCanceling(
             ExecutionGraph executionGraph,

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java

@@ -92,6 +92,24 @@ public void testTransitionToSubsequentStateWhenResourceChanged() throws Exceptio
         }
     }
 
+    @Test
+    public void testTransitionToSubsequentStateWhenResourceChangedButHasDesiredResources()
+            throws Exception {
+        try (MockRestartingContext ctx = new MockRestartingContext()) {
+            JobVertexID jobVertexId = new JobVertexID();
+            VertexParallelism availableParallelism =
+                    new VertexParallelism(singletonMap(jobVertexId, 1));
+            VertexParallelism requiredParallelismForForcedRestart =
+                    new VertexParallelism(singletonMap(jobVertexId, 2));
+
+            ctx.setAvailableVertexParallelism(availableParallelism);
+            ctx.setHasDesiredResources(true);
+            Restarting restarting = createRestartingState(ctx, requiredParallelismForForcedRestart);
+            ctx.setExpectCreatingExecutionGraph();
+            restarting.onGloballyTerminalState(JobStatus.CANCELED);
+        }
+    }
+
     @Test
     void testCancel() throws Exception {
         try (MockRestartingContext ctx = new MockRestartingContext()) {