fix

JunRuiLee · Feb 8, 2025 · 8b406bf · 8b406bf
1 parent 4e2bc4d
commit 8b406bf
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 13 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java
@@ -71,7 +71,7 @@ public class StreamEdge implements Serializable {
     /** The name of the operator in the target vertex. */
     private final String targetOperatorName;
 
-    private final StreamExchangeMode exchangeMode;
+    private StreamExchangeMode exchangeMode;
 
     private long bufferTimeout;
 
@@ -195,6 +195,10 @@ public StreamExchangeMode getExchangeMode() {
         return exchangeMode;
     }
 
+    public void setExchangeMode(StreamExchangeMode exchangeMode) {
+        this.exchangeMode = exchangeMode;
+    }
+
     public void setPartitioner(StreamPartitioner<?> partitioner) {
         configureKeyCorrelation(partitioner);
         this.outputPartitioner = partitioner;

diff --git a/...untime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/...untime/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -829,9 +829,7 @@ private void setVertexParallelismsForDynamicGraphIfNecessary() {
         jobVertices.forEach(
                 (startNodeId, jobVertex) -> {
                     Set<JobVertex> forwardConsumers =
-                            jobVertexBuildContext
-                                    .getChainInfo(startNodeId)
-                                    .getTransitiveOutEdges()
+                            jobVertexBuildContext.getChainInfo(startNodeId).getTransitiveOutEdges()
                                     .stream()
                                     .filter(
                                             edge ->
@@ -1533,7 +1531,7 @@ private static boolean allHybridOrSameReconsumablePartitionType(
                         && partitionType2.isHybridResultPartition());
     }
 
-    public static void tryConvertPartitionerForDynamicGraph(
+    private static void tryConvertPartitionerForDynamicGraph(
             List<StreamEdge> chainableOutputs,
             List<StreamEdge> nonChainableOutputs,
             JobVertexBuildContext jobVertexBuildContext) {
@@ -2113,11 +2111,9 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
                 groupOperatorIds.stream()
                         .flatMap(
                                 (oid) ->
-                                        streamGraph
-                                                .getStreamNode(oid)
+                                        streamGraph.getStreamNode(oid)
                                                 .getManagedMemoryOperatorScopeUseCaseWeights()
-                                                .entrySet()
-                                                .stream())
+                                                .entrySet().stream())
                         .collect(
                                 Collectors.groupingBy(
                                         Map.Entry::getKey,
@@ -2127,10 +2123,8 @@ private static void setManagedMemoryFractionForSlotSharingGroup(
                 groupOperatorIds.stream()
                         .flatMap(
                                 (oid) ->
-                                        streamGraph
-                                                .getStreamNode(oid)
-                                                .getManagedMemorySlotScopeUseCases()
-                                                .stream())
+                                        streamGraph.getStreamNode(oid)
+                                                .getManagedMemorySlotScopeUseCases().stream())
                         .collect(Collectors.toSet());
 
         for (JobVertexID jobVertexID : jobVertexIds) {