1996fanrui commented on code in PR #29064:
URL: https://github.com/apache/flink/pull/29064#discussion_r3936382607
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java:
##########
@@ -221,7 +225,48 @@ private static <T> GateFilterHandler<T> createGateHandler(
return null;
}
- return new GateFilterHandler<>(gateVirtualChannels, elementSerializer);
+ return new GateFilterHandler<>(gateVirtualChannels, elementSerializer,
channelMapping);
+ }
+
+ /**
+ * Maps each old-channel key to the virtual channels that fold into its
new channel, so
+ * watermarks/statuses can be aggregated across old channels merged on a
fan-in rescale. Without
+ * a rescale each group is a singleton and aggregation is verbatim
pass-through.
+ */
+ private static <T>
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>>
buildWatermarkMergeGroups(
+ Map<SubtaskConnectionDescriptor, VirtualChannel<T>>
gateVirtualChannels,
+ RescaleMappings channelMapping) {
+ RescaleMappings oldToNewMapping = channelMapping.invert();
+ Map<Integer, List<VirtualChannel<T>>> byNewChannel = new HashMap<>();
+ Map<SubtaskConnectionDescriptor, Integer> keyToNewChannel = new
HashMap<>();
+ gateVirtualChannels.forEach(
+ (key, vc) -> {
+ int newChannelIndex = newChannelIndexOf(key,
oldToNewMapping);
+ byNewChannel.computeIfAbsent(newChannelIndex, k -> new
ArrayList<>()).add(vc);
+ keyToNewChannel.put(key, newChannelIndex);
+ });
+
+ Map<SubtaskConnectionDescriptor, List<VirtualChannel<T>>> mergeGroups
= new HashMap<>();
+ keyToNewChannel.forEach(
+ (key, newChannelIndex) -> mergeGroups.put(key,
byNewChannel.get(newChannelIndex)));
+ return mergeGroups;
Review Comment:
Nice, and simplified!
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
