[hotfix] [streaming] Correct copying StreamPartitioner in rescale case

apache · May 31, 2024 · e095657 · e095657
1 parent 90a71a1
commit e095657
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/.../java/org/apache/flink/streaming/runtime/io/recovery/RescalingStreamTaskNetworkInput.java b/.../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);
         }
 
         private StreamPartitioner<T> createPartitioner(Integer index) {