ldadima commented on code in PR #24784:
URL: https://github.com/apache/flink/pull/24784#discussion_r1606219365
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java:
##########
@@ -232,7 +232,9 @@ public Predicate<StreamRecord<T>> apply(InputChannelInfo
channelInfo) {
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);
Review Comment:
Extracted to `[hotfix]` [PR](https://github.com/apache/flink/pull/24811)
Can you also review this one too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
