[FLINK-36066][runtime] Introduce the AdaptiveGraphManager component

flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/IntermediateDataSet.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.jobgraph;
 
 import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
+import org.apache.flink.streaming.api.graph.StreamEdge;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -43,6 +44,8 @@ public class IntermediateDataSet implements java.io.Serializable {
     // All consumers must have the same partitioner and parallelism
     private final List<JobEdge> consumers = new ArrayList<>();
 
+    private final List<StreamEdge> outputStreamEdges = new ArrayList<>();
+
     // The type of partition to use at runtime
     private final ResultPartitionType resultType;
 
@@ -85,13 +88,17 @@ public ResultPartitionType getResultType() {
         return resultType;
     }
 
+    public List<StreamEdge> getOutputStreamEdges() {
+        return outputStreamEdges;
+    }
+
     // --------------------------------------------------------------------------------------------
 
     public void addConsumer(JobEdge edge) {
         // sanity check
         checkState(id.equals(edge.getSourceId()), "Incompatible dataset id.");
 
-        if (consumers.isEmpty()) {
+        if (consumers.isEmpty() && outputStreamEdges.isEmpty()) {
             distributionPattern = edge.getDistributionPattern();
             isBroadcast = edge.isBroadcast();
         } else {
@@ -103,6 +110,23 @@ public void addConsumer(JobEdge edge) {
         consumers.add(edge);
     }
 
+    public void addOutputStreamEdge(StreamEdge edge) {
+        checkState(consumers.isEmpty(), "The output job edges have already been added.");
+        DistributionPattern pattern =
+                edge.getPartitioner().isPointwise()
+                        ? DistributionPattern.POINTWISE
+                        : DistributionPattern.ALL_TO_ALL;
+        if (outputStreamEdges.isEmpty()) {
+            distributionPattern = pattern;
+            isBroadcast = edge.getPartitioner().isBroadcast();
+        } else {
+            checkState(distributionPattern == pattern, "Incompatible distribution pattern.");
+            checkState(
+                    isBroadcast == edge.getPartitioner().isBroadcast(),
+                    "Incompatible broadcast type.");
+        }
+        outputStreamEdges.add(edge);
+    }
     // --------------------------------------------------------------------------------------------
 
     @Override

flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphGenerator.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.graph;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+import java.util.List;
+
+/**
+ * Defines the mechanism for dynamically adapting the graph topology of a Flink job at runtime. The
+ * AdaptiveGraphGenerator is responsible for managing and updating the job's execution plan based on
+ * runtime events such as the completion of a job vertex. It provides functionalities to generate
+ * new job vertices, retrieve the current JobGraph, update the StreamGraph, and track the status of
+ * pending stream nodes.
+ */
+@Internal
+public interface AdaptiveGraphGenerator {
+
+    /**
+     * Retrieves the JobGraph representation of the current state of the Flink job.
+     *
+     * @return The {@link JobGraph} instance.
+     */
+    JobGraph getJobGraph();
+
+    /**
+     * Retrieves the StreamGraphContext which provides a read-only view of the StreamGraph and
+     * methods to modify its StreamEdges and StreamNodes.
+     *
+     * @return an instance of {@link StreamGraphContext}.
+     */
+    StreamGraphContext getStreamGraphContext();
+
+    /**
+     * Responds to notifications that a JobVertex has completed execution. This method generates new
+     * job vertices, incorporates them into the JobGraph, and returns a list of the newly created
+     * JobVertex instances.
+     *
+     * @param finishedJobVertexId The ID of the completed JobVertex.
+     * @return A list of the newly added {@link JobVertex} instances to the JobGraph.
+     */
+    List<JobVertex> onJobVertexFinished(JobVertexID finishedJobVertexId);
+}