You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2024/05/31 08:00:08 UTC
(flink) 01/03: [hotfix] [streaming] Correct copying StreamPartitioner in rescale case
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e0956578e1dfe5f3c54c3a00691d9603b673a2ec
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) {