[flink] branch release-1.16 updated: [FLINK-29576][runtime] Collects serialized OperatorCoordinators before adding them to JobVertex

[mapohl]

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

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


The following commit(s) were added to refs/heads/release-1.16 by this push:
     new b05fac1560a [FLINK-29576][runtime] Collects serialized 
OperatorCoordinators before adding them to JobVertex
b05fac1560a is described below

commit b05fac1560add86dcb41b0b5a8297326dbc99a19
Author: Matthias Pohl <matthias.p...@aiven.io>
AuthorDate: Tue Oct 11 13:07:52 2022 +0200

    [FLINK-29576][runtime] Collects serialized OperatorCoordinators before 
adding them to JobVertex
    
    The serialization was parallelized in FLINK-26675. Adding the 
OperatorCoordinator
    instances concurrently occasionally caused problems with instances being 
dropped.
---
 .../api/graph/StreamingJobGraphGenerator.java      | 40 ++++++++++++++++++----
 1 file changed, 34 insertions(+), 6 deletions(-)

diff --git 
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
 
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index a3a173a5d50..dead3aac68d 100644
--- 
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ 
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -105,6 +105,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -184,7 +185,10 @@ public class StreamingJobGraphGenerator {
     private final Executor serializationExecutor;
 
     // Futures for the serialization of operator coordinators
-    private final List<CompletableFuture<Void>> 
coordinatorSerializationFutures = new ArrayList<>();
+    private final Map<
+                    JobVertexID,
+                    
List<CompletableFuture<SerializedValue<OperatorCoordinator.Provider>>>>
+            coordinatorSerializationFuturesPerJobVertex = new HashMap<>();
 
     private final Map<Integer, Map<StreamEdge, NonChainedOutput>> 
opIntermediateOutputs;
 
@@ -286,7 +290,8 @@ public class StreamingJobGraphGenerator {
                                                             
serializationExecutor))
                                     .collect(Collectors.toList()))
                     .get();
-            FutureUtils.combineAll(coordinatorSerializationFutures).get();
+
+            waitForSerializationFuturesAndUpdateJobVertices();
         } catch (Exception e) {
             throw new FlinkRuntimeException("Error in serialization.", e);
         }
@@ -298,6 +303,25 @@ public class StreamingJobGraphGenerator {
         return jobGraph;
     }
 
+    private void waitForSerializationFuturesAndUpdateJobVertices()
+            throws ExecutionException, InterruptedException {
+        for (Map.Entry<
+                        JobVertexID,
+                        
List<CompletableFuture<SerializedValue<OperatorCoordinator.Provider>>>>
+                futuresPerJobVertex : 
coordinatorSerializationFuturesPerJobVertex.entrySet()) {
+            final JobVertexID jobVertexId = futuresPerJobVertex.getKey();
+            final JobVertex jobVertex = jobGraph.findVertexByID(jobVertexId);
+
+            Preconditions.checkState(
+                    jobVertex != null,
+                    "OperatorCoordinator providers were registered for 
JobVertexID '%s' but no corresponding JobVertex can be found.",
+                    jobVertexId);
+            FutureUtils.combineAll(futuresPerJobVertex.getValue())
+                    .get()
+                    .forEach(jobVertex::addOperatorCoordinator);
+        }
+    }
+
     private void addVertexIndexPrefixInVertexName() {
         if (!streamGraph.isVertexNameIncludeIndexPrefix()) {
             return;
@@ -823,14 +847,15 @@ public class StreamingJobGraphGenerator {
             
jobVertex.addIntermediateDataSetIdToConsume(streamNode.getConsumeClusterDatasetId());
         }
 
+        final 
List<CompletableFuture<SerializedValue<OperatorCoordinator.Provider>>>
+                serializationFutures = new ArrayList<>();
         for (OperatorCoordinator.Provider coordinatorProvider :
                 chainInfo.getCoordinatorProviders()) {
-            coordinatorSerializationFutures.add(
-                    CompletableFuture.runAsync(
+            serializationFutures.add(
+                    CompletableFuture.supplyAsync(
                             () -> {
                                 try {
-                                    jobVertex.addOperatorCoordinator(
-                                            new 
SerializedValue<>(coordinatorProvider));
+                                    return new 
SerializedValue<>(coordinatorProvider);
                                 } catch (IOException e) {
                                     throw new FlinkRuntimeException(
                                             String.format(
@@ -841,6 +866,9 @@ public class StreamingJobGraphGenerator {
                             },
                             serializationExecutor));
         }
+        if (!serializationFutures.isEmpty()) {
+            coordinatorSerializationFuturesPerJobVertex.put(jobVertexId, 
serializationFutures);
+        }
 
         jobVertex.setResources(
                 chainedMinResources.get(streamNodeId), 
chainedPreferredResources.get(streamNodeId));