1996fanrui commented on code in PR #28651:
URL: https://github.com/apache/flink/pull/28651#discussion_r3547681836
##########
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java:
##########
@@ -73,28 +73,156 @@ void recover(Info info, int oldSubtaskIndex,
BufferWithContext<Context> bufferWi
throws IOException, InterruptedException;
}
-class InputChannelRecoveredStateHandler
+/**
+ * Abstract base for all input-channel recovery handlers. Holds the channel
mapping logic shared by
+ * both variants (no-filtering, filtering).
+ *
+ * <p>Subclasses implement {@link #recover} according to their specific
recovery mode and override
+ * {@link #closeInternal()} to release mode-specific resources.
+ *
+ * <p>Use the static {@link #create} factory to obtain the correct concrete
subclass.
+ */
+abstract class AbstractInputChannelRecoveredStateHandler
implements RecoveredChannelStateHandler<InputChannelInfo, Buffer> {
- private final InputGate[] inputGates;
- private final InflightDataRescalingDescriptor channelMapping;
+ final InputGate[] inputGates;
+ final InflightDataRescalingDescriptor channelMapping;
+ final Map<InputChannelInfo, RecoveredInputChannel> rescaledChannels = new
HashMap<>();
+ final Map<Integer, RescaleMappings> oldToNewMappings = new HashMap<>();
- private final Map<InputChannelInfo, RecoveredInputChannel>
rescaledChannels = new HashMap<>();
- private final Map<Integer, RescaleMappings> oldToNewMappings = new
HashMap<>();
+ AbstractInputChannelRecoveredStateHandler(
+ InputGate[] inputGates, InflightDataRescalingDescriptor
channelMapping) {
+ this.inputGates = inputGates;
+ this.channelMapping = channelMapping;
+ }
/**
- * Optional filtering handler for filtering recovered buffers. When
non-null, filtering is
- * performed during recovery in the channel-state-unspilling thread.
+ * Factory that selects the correct subclass based on {@code
checkpointingDuringRecoveryEnabled}
+ * and whether a {@code filteringHandler} is present.
+ *
+ * <ul>
+ * <li>{@code false} → {@link NoSpillingHandler}
+ * <li>{@code true} and {@code filteringHandler == null} → {@link
NoSpillingHandler}
+ * <li>{@code true} and {@code filteringHandler != null} → {@link
FilteringHandler}
+ * </ul>
*/
- @Nullable private final ChannelStateFilteringHandler filteringHandler;
+ static AbstractInputChannelRecoveredStateHandler create(
+ InputGate[] inputGates,
+ InflightDataRescalingDescriptor channelMapping,
+ boolean checkpointingDuringRecoveryEnabled,
+ @Nullable ChannelStateFilteringHandler filteringHandler,
+ int memorySegmentSize) {
+ if (!checkpointingDuringRecoveryEnabled) {
+ return new NoSpillingHandler(inputGates, channelMapping);
+ }
+ // FLINK-38544 transitional: the flag-on path still uses the in-memory
handlers until the
+ // spilling backend lands.
+ if (filteringHandler == null) {
+ return new NoSpillingHandler(inputGates, channelMapping);
+ }
+ return new FilteringHandler(
+ inputGates, channelMapping, filteringHandler,
memorySegmentSize);
+ }
Review Comment:
Hey, thanks for the review!
Good question, it is an existing design on CDR v1.
`VirtualChannel` is the deserialize+filter primitive keyed by the old
(pre-rescale) channel, and recovered state is stored per old channel — so even
with CDR we still need it to filter the pre-filter data; we just moved it from
the task thread into the recovery/filtering layer
(ChannelStateFilteringHandler). The redundant task-thread copy is skipped via
`StreamTaskNetworkInputFactory` (returns plain `StreamTaskNetworkInput` when
CDR is on).
--
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]
