This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch release-1.18 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 949a5551ad4339279b112ec734a2f036d1e5b34e Author: Dmitriy Linevich <[email protected]> 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) {
