rkhachatryan commented on code in PR #28651:
URL: https://github.com/apache/flink/pull/28651#discussion_r3546138887
##########
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:
In case of rescaling:
- without CDR, we perform filtering somewhere in VirtualChannel
- with CDR, we perform filtering before delivering to InputChannels
So, with CDR we should not create create VirtualChannel s, right?
Did we change that path (in some future PR)?
--
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]
