(flink) 01/02: [hotfix][runtime] Fix the intra-input correlation of ForwardForUnspecifiedPartitioner

[ju...@apache.org]

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

junrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 3790af913214f94c658ad36669c6fd185a9be010
Author: noorall <86...@qq.com>
AuthorDate: Wed Feb 12 19:08:19 2025 +0800

    [hotfix][runtime] Fix the intra-input correlation of ForwardForUnspecifiedPartitioner
---
 .../main/java/org/apache/flink/streaming/api/graph/StreamEdge.java   | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

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
index 0997f180a29..146b432b7ff 100644
--- 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
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.api.graph;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
+import org.apache.flink.streaming.runtime.partitioner.ForwardForUnspecifiedPartitioner;
 import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
 import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
 import org.apache.flink.streaming.runtime.partitioner.StreamPartitioner;
@@ -277,7 +278,9 @@ public class StreamEdge implements Serializable {
         // work normally by default. The final value of the correlations can be flexibly determined
         // by the operator.
         if (partitioner.isPointwise()) {
-            this.intraInputKeyCorrelated = partitioner instanceof ForwardPartitioner;
+            this.intraInputKeyCorrelated =
+                    partitioner instanceof ForwardPartitioner
+                            && !(partitioner instanceof ForwardForUnspecifiedPartitioner);
             this.interInputsKeysCorrelated = false;
         } else {
             this.intraInputKeyCorrelated = !(partitioner instanceof RebalancePartitioner);