(flink) branch master updated: [hotfix] [streaming] Correct copying StreamPartitioner in rescale case

[pn...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 11bd030aca8 [hotfix] [streaming] Correct copying StreamPartitioner in rescale case
11bd030aca8 is described below

commit 11bd030aca8bc7134c6ff6893919a2cb0eaf86bf
Author: Dmitriy Linevich <ld...@mail.ru>
AuthorDate: Mon May 20 11:05:21 2024 +0700

    [hotfix] [streaming] Correct copying StreamPartitioner in rescale case
---
 .../runtime/io/recovery/RescalingStreamTaskNetworkInput.java          | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java
index 7b8002eee2a..89089512cba 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java
@@ -232,7 +232,9 @@ public final class RescalingStreamTaskNetworkInput<T>
                             channelInfo.getGateIdx(), this::createPartitioner);
             // use a copy of partitioner to ensure that the filter of ambiguous virtual channels
             // have the same state across several subtasks
-            return new RecordFilter<>(partitioner.copy(), inputSerializer, subtaskIndex);
+            StreamPartitioner<T> partitionerCopy = partitioner.copy();
+            partitionerCopy.setup(numberOfChannels);
+            return new RecordFilter<>(partitionerCopy, inputSerializer, subtaskIndex);
         }
 
         private StreamPartitioner<T> createPartitioner(Integer index) {