(flink) branch release-1.19 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.19
in repository https://gitbox.apache.org/repos/asf/flink.git


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

commit 87ed9ccc2103457ba91f6ca45adfd2bfcc75c9ac
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 5f6438ce181..238c594fd55 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
@@ -1187,7 +1187,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 da90ef1468d..e9b1317e46e 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
@@ -123,7 +123,6 @@ public class CreatingExecutionGraph extends 
StateWithoutExecutionGraph {
                         
operatorCoordinatorHandlerFactory.create(executionGraph, context);
                 operatorCoordinatorHandler.initializeOperatorCoordinators(
                         context.getMainThreadExecutor());
-                operatorCoordinatorHandler.startAllOperatorCoordinators();
                 final String updatedPlan =
                         JsonPlanGenerator.generatePlan(
                                 executionGraph.getJobID(),
@@ -137,6 +136,10 @@ public class CreatingExecutionGraph extends 
StateWithoutExecutionGraph {
                                                 .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 b831b3bb62f..0f89cdf7e12 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
@@ -93,8 +93,12 @@ class CreatingExecutionGraphTest {
                 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