[FLINK-35159] Transition ExecutionGraph to RUNNING after slot assignment

twalthr · Apr 19, 2024 · 87ed9cc · 87ed9cc
1 parent a2c3d27
commit 87ed9cc
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/...-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -1187,7 +1187,6 @@ public CreatingExecutionGraph.AssignmentResult tryToAssignSlots(
                 executionGraphWithVertexParallelism.getExecutionGraph();
 
         executionGraph.start(componentMainThreadExecutor);
-        executionGraph.transitionToRunning();
 
         executionGraph.setInternalTaskFailuresListener(
                 new UpdateSchedulerNgOnInternalFailuresListener(this));

diff --git a/...ime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java b/...ime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
@@ -123,7 +123,6 @@ private void handleExecutionGraphCreation(
                         operatorCoordinatorHandlerFactory.create(executionGraph, context);
                 operatorCoordinatorHandler.initializeOperatorCoordinators(
                         context.getMainThreadExecutor());
-                operatorCoordinatorHandler.startAllOperatorCoordinators();
                 final String updatedPlan =
                         JsonPlanGenerator.generatePlan(
                                 executionGraph.getJobID(),
@@ -137,6 +136,10 @@ private void handleExecutionGraphCreation(
                                                 .iterator(),
                                 executionGraphWithVertexParallelism.getVertexParallelism());
                 executionGraph.setJsonPlan(updatedPlan);
+
+                executionGraph.transitionToRunning();
+                operatorCoordinatorHandler.startAllOperatorCoordinators();
+
                 context.goToExecuting(
                         result.getExecutionGraph(),
                         executionGraphHandler,

diff --git a/...src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/...src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
@@ -93,8 +93,12 @@ void testNotPossibleSlotAssignmentTransitionsToWaitingForResources() {
                 ignored -> CreatingExecutionGraph.AssignmentResult.notPossible());
         context.setExpectWaitingForResources();
 
-        executionGraphWithVertexParallelismFuture.complete(
-                getGraph(new StateTrackingMockExecutionGraph()));
+        final StateTrackingMockExecutionGraph executionGraph =
+                new StateTrackingMockExecutionGraph();
+
+        executionGraphWithVertexParallelismFuture.complete(getGraph(executionGraph));
+
+        assertThat(executionGraph.getState()).isEqualTo(JobStatus.INITIALIZING);
     }
 
     @Test