(flink) branch release-1.18 updated: [FLINK-35159] Transition ExecutionGraph to RUNNING after slot assignment

[chesnay]

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

chesnay pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.18 by this push:
     new aacc735806a [FLINK-35159] Transition ExecutionGraph to RUNNING after 
slot assignment
aacc735806a is described below

commit aacc735806acf1d63fa732706e079bc2ca1bb4fc
Author: Chesnay Schepler <ches...@apache.org>
AuthorDate: Thu Apr 18 19:10:42 2024 +0200

    [FLINK-35159] Transition ExecutionGraph to RUNNING after slot assignment
---
 .../flink/runtime/scheduler/adaptive/AdaptiveScheduler.java       | 1 -
 .../flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java  | 5 ++++-
 .../runtime/scheduler/adaptive/CreatingExecutionGraphTest.java    | 8 ++++++--
 3 files changed, 10 insertions(+), 4 deletions(-)

diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 4ee22c95848..34539d23e04 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -1096,7 +1096,6 @@ public class AdaptiveScheduler
                 executionGraphWithVertexParallelism.getExecutionGraph();
 
         executionGraph.start(componentMainThreadExecutor);
-        executionGraph.transitionToRunning();
 
         executionGraph.setInternalTaskFailuresListener(
                 new UpdateSchedulerNgOnInternalFailuresListener(this));
diff --git 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
index c876fe6ad1d..21055945372 100644
--- 
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
+++ 
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
@@ -124,7 +124,6 @@ public class CreatingExecutionGraph implements State {
                         
operatorCoordinatorHandlerFactory.create(executionGraph, context);
                 operatorCoordinatorHandler.initializeOperatorCoordinators(
                         context.getMainThreadExecutor());
-                operatorCoordinatorHandler.startAllOperatorCoordinators();
                 final String updatedPlan =
                         JsonPlanGenerator.generatePlan(
                                 executionGraph.getJobID(),
@@ -138,6 +137,10 @@ public class CreatingExecutionGraph implements State {
                                                 .iterator(),
                                 
executionGraphWithVertexParallelism.getVertexParallelism());
                 executionGraph.setJsonPlan(updatedPlan);
+
+                executionGraph.transitionToRunning();
+                operatorCoordinatorHandler.startAllOperatorCoordinators();
+
                 context.goToExecuting(
                         result.getExecutionGraph(),
                         executionGraphHandler,
diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
index 2375a194206..69e5f589b19 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
@@ -157,8 +157,12 @@ public class CreatingExecutionGraphTest extends TestLogger 
{
                     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);
         }
     }