This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0076de581afcd72367d94ecba5e9981524701ff9 Author: Rui Fan <[email protected]> AuthorDate: Sun Jul 5 23:59:10 2026 +0200 [FLINK-39520][checkpoint] Extract AbstractInputChannelRecoveredStateHandler with concrete no-filtering/filtering handlers Single-file class split of RecoveredChannelStateHandler.java; all classes stay package-private in this file. No behavior change. - New abstract AbstractInputChannelRecoveredStateHandler: fields inputGates, channelMapping, rescaledChannels, oldToNewMappings; methods getMappedChannels, calculateMapping, getChannel - verbatim extraction from InputChannelRecoveredStateHandler. Adds a closeInternal() template hook used by later phases. - New NoSpillingHandler: recover() is the verbatim old non-filtering branch (onRecoveredStateBuffer(descriptor) followed by onRecoveredStateBuffer(buffer.retainBuffer()), same try/finally recycle); inherits the verbatim network-pool getBuffer. - New FilteringHandler (the v1 filtering behavior, verbatim): recover() = old recoverWithFiltering (filteringHandler.filterAndRewrite(..., channel::requestBufferBlocking) delivering List<Buffer> via onRecoveredStateBuffer); getBuffer = old getPreFilterBuffer (reusable heap MemorySegment, preFilterBufferInUse invariant); closeInternal = old segment-freeing close. - New static create(...) factory selecting NoSpillingHandler vs FilteringHandler - equivalent to the old internal if-branch. SequentialChannelStateReaderImpl switches from direct construction to the factory (mechanical). - ResultSubpartitionRecoveredStateHandler: untouched. - Trace label in RecoveredInputChannel.onRecoveredStateBuffer: "InputChannelRecoveredStateHandler#recover" -> "NoSpillingHandler#recover" (final label). - Tests: mechanical adaptation of InputChannelRecoveredStateHandlerTest, RecoveredChannelStateHandlerTest. --- .../channel/RecoveredChannelStateHandler.java | 241 ++++++++++++++------- .../channel/SequentialChannelStateReaderImpl.java | 5 +- .../partition/consumer/RecoveredInputChannel.java | 5 +- .../InputChannelRecoveredStateHandlerTest.java | 67 +++--- .../channel/RecoveredChannelStateHandlerTest.java | 2 +- 5 files changed, 197 insertions(+), 123 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java index ca01ff37bd3..8f1eeb24fec 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java @@ -73,28 +73,156 @@ interface RecoveredChannelStateHandler<Info, Context> extends AutoCloseable { 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); + } + + /** Default buffer allocation from the network buffer pool, used by non-filtering modes. */ + @Override + public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo) + throws IOException, InterruptedException { + RecoveredInputChannel channel = getMappedChannels(channelInfo); + Buffer buffer = channel.requestBufferBlocking(); + return new BufferWithContext<>(wrap(buffer), buffer); + } + + @Override + public void close() throws IOException { + // note that we need to finish all RecoveredInputChannels, not just those with state + for (final InputGate inputGate : inputGates) { + inputGate.finishReadRecoveredState(); + } + closeInternal(); + } + + /** Hook for subclasses to release their own resources. Called by {@link #close()}. */ + void closeInternal() throws IOException {} + + RecoveredInputChannel getMappedChannels(InputChannelInfo channelInfo) { + return rescaledChannels.computeIfAbsent(channelInfo, this::calculateMapping); + } + + @Nonnull + private RecoveredInputChannel calculateMapping(InputChannelInfo info) { + final RescaleMappings oldToNewMapping = + oldToNewMappings.computeIfAbsent( + info.getGateIdx(), idx -> channelMapping.getChannelMapping(idx).invert()); + int[] mappedIndexes = oldToNewMapping.getMappedIndexes(info.getInputChannelIdx()); + checkState( + mappedIndexes.length == 1, + "One buffer is only distributed to one target InputChannel since " + + "one buffer is expected to be processed once by the same task."); + return getChannel(info.getGateIdx(), mappedIndexes[0]); + } + + private RecoveredInputChannel getChannel(int gateIndex, int subPartitionIndex) { + final InputChannel inputChannel = inputGates[gateIndex].getChannel(subPartitionIndex); + if (!(inputChannel instanceof RecoveredInputChannel)) { + throw new IllegalStateException( + "Cannot restore state to a non-recovered input channel: " + inputChannel); + } + return (RecoveredInputChannel) inputChannel; + } +} + +/** + * Recovery handler for the case where checkpointing during recovery is disabled. Delivers recovered + * buffers directly into the input channel via {@code onRecoveredStateBuffer}, with no spill file. + */ +class NoSpillingHandler extends AbstractInputChannelRecoveredStateHandler { + + NoSpillingHandler(InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) { + super(inputGates, channelMapping); + } + + @Override + public void recover( + InputChannelInfo channelInfo, + int oldSubtaskIndex, + BufferWithContext<Buffer> bufferWithContext) + throws IOException, InterruptedException { + Buffer buffer = bufferWithContext.context; + try { + if (buffer.readableBytes() > 0) { + RecoveredInputChannel channel = getMappedChannels(channelInfo); + channel.onRecoveredStateBuffer( + EventSerializer.toBuffer( + new SubtaskConnectionDescriptor( + oldSubtaskIndex, channelInfo.getInputChannelIdx()), + false)); + channel.onRecoveredStateBuffer(buffer.retainBuffer()); + } + } finally { + buffer.recycleBuffer(); + } + } +} + +/** + * Recovery handler for the case where checkpointing during recovery is enabled and a filtering + * handler is present. Uses a reusable heap-backed pre-filter buffer (isolated from the Network + * Buffer Pool), filters recovered buffers through {@link + * ChannelStateFilteringHandler#filterAndRewrite}, and delivers the filtered buffers directly into + * the input channel via {@code onRecoveredStateBuffer}. + */ +class FilteringHandler extends AbstractInputChannelRecoveredStateHandler { + + private final ChannelStateFilteringHandler filteringHandler; /** Network buffer memory segment size in bytes. Used to size the reusable pre-filter buffer. */ private final int memorySegmentSize; /** * Reusable heap memory segment backing the pre-filter buffer in filtering mode. Lazily - * allocated on the first {@link #getPreFilterBuffer} call, reused for every subsequent call, - * and freed in {@link #close()}. + * allocated on the first {@link #getBuffer} call, reused for every subsequent call, and freed + * in {@link #closeInternal()}. * * <p>Reuse is safe because at most one pre-filter buffer is in flight per task at any moment. * This invariant is enforced at runtime by {@link #preFilterBufferInUse}. @@ -108,39 +236,26 @@ class InputChannelRecoveredStateHandler */ private boolean preFilterBufferInUse; - InputChannelRecoveredStateHandler( + FilteringHandler( InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping, - @Nullable ChannelStateFilteringHandler filteringHandler, + ChannelStateFilteringHandler filteringHandler, int memorySegmentSize) { - this.inputGates = inputGates; - this.channelMapping = channelMapping; + super(inputGates, channelMapping); this.filteringHandler = filteringHandler; checkArgument( memorySegmentSize > 0, "memorySegmentSize must be positive: %s", memorySegmentSize); this.memorySegmentSize = memorySegmentSize; } - @Override - public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo) - throws IOException, InterruptedException { - if (filteringHandler != null) { - return getPreFilterBuffer(); - } - // Non-filtering mode: use existing network buffer pool allocation. - RecoveredInputChannel channel = getMappedChannels(channelInfo); - Buffer buffer = channel.requestBufferBlocking(); - return new BufferWithContext<>(wrap(buffer), buffer); - } - /** * Allocates a pre-filter buffer from a reusable heap segment (isolated from the Network Buffer * Pool) in filtering mode. * * <p>Memory management: a single {@link MemorySegment} per task is lazily allocated on first * invocation and reused across every subsequent call. The custom {@link BufferRecycler} does - * not free the segment — it only flips {@link #preFilterBufferInUse} back to {@code false} so - * the next call can reuse it. The segment itself is freed in {@link #close()}. + * not free the segment; it only flips {@link #preFilterBufferInUse} back to {@code false} so + * the next call can reuse it. The segment itself is freed in {@link #closeInternal()}. * * <p>Runtime invariant check: the one-at-a-time invariant on pre-filter buffers is guaranteed * by Flink's serial recovery loop and the deserializer's ownership contract. This method @@ -148,7 +263,8 @@ class InputChannelRecoveredStateHandler * recycled, it throws {@link IllegalStateException} so any future regression fails loudly * instead of silently corrupting memory. */ - private BufferWithContext<Buffer> getPreFilterBuffer() { + @Override + public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo) { checkState( !preFilterBufferInUse, "Previous pre-filter buffer has not been recycled. This violates the " @@ -165,17 +281,6 @@ class InputChannelRecoveredStateHandler return new BufferWithContext<>(wrap(buffer), buffer); } - @VisibleForTesting - boolean isPreFilterBufferInUse() { - return preFilterBufferInUse; - } - - @VisibleForTesting - @Nullable - MemorySegment getPreFilterSegmentForTesting() { - return preFilterSegment; - } - @Override public void recover( InputChannelInfo channelInfo, @@ -186,18 +291,7 @@ class InputChannelRecoveredStateHandler try { if (buffer.readableBytes() > 0) { RecoveredInputChannel channel = getMappedChannels(channelInfo); - - if (filteringHandler != null) { - recoverWithFiltering( - channel, channelInfo, oldSubtaskIndex, buffer.retainBuffer()); - } else { - channel.onRecoveredStateBuffer( - EventSerializer.toBuffer( - new SubtaskConnectionDescriptor( - oldSubtaskIndex, channelInfo.getInputChannelIdx()), - false)); - channel.onRecoveredStateBuffer(buffer.retainBuffer()); - } + recoverWithFiltering(channel, channelInfo, oldSubtaskIndex, buffer.retainBuffer()); } } finally { buffer.recycleBuffer(); @@ -232,44 +326,25 @@ class InputChannelRecoveredStateHandler } } + @VisibleForTesting + boolean isPreFilterBufferInUse() { + return preFilterBufferInUse; + } + + @VisibleForTesting + @Nullable + MemorySegment getPreFilterSegmentForTesting() { + return preFilterSegment; + } + @Override - public void close() throws IOException { - // note that we need to finish all RecoveredInputChannels, not just those with state - for (final InputGate inputGate : inputGates) { - inputGate.finishReadRecoveredState(); - } + void closeInternal() throws IOException { if (preFilterSegment != null) { preFilterSegment.free(); preFilterSegment = null; preFilterBufferInUse = false; } } - - private RecoveredInputChannel getChannel(int gateIndex, int subPartitionIndex) { - final InputChannel inputChannel = inputGates[gateIndex].getChannel(subPartitionIndex); - if (!(inputChannel instanceof RecoveredInputChannel)) { - throw new IllegalStateException( - "Cannot restore state to a non-recovered input channel: " + inputChannel); - } - return (RecoveredInputChannel) inputChannel; - } - - private RecoveredInputChannel getMappedChannels(InputChannelInfo channelInfo) { - return rescaledChannels.computeIfAbsent(channelInfo, this::calculateMapping); - } - - @Nonnull - private RecoveredInputChannel calculateMapping(InputChannelInfo info) { - final RescaleMappings oldToNewMapping = - oldToNewMappings.computeIfAbsent( - info.getGateIdx(), idx -> channelMapping.getChannelMapping(idx).invert()); - int[] mappedIndexes = oldToNewMapping.getMappedIndexes(info.getInputChannelIdx()); - checkState( - mappedIndexes.length == 1, - "One buffer is only distributed to one target InputChannel since " - + "one buffer is expected to be processed once by the same task."); - return getChannel(info.getGateIdx(), mappedIndexes[0]); - } } class ResultSubpartitionRecoveredStateHandler diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java index c52572e52fa..3ebad6eb3a2 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java @@ -70,10 +70,11 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR : null; try (ChannelStateFilteringHandler ignored = filteringHandler; - InputChannelRecoveredStateHandler stateHandler = - new InputChannelRecoveredStateHandler( + AbstractInputChannelRecoveredStateHandler stateHandler = + AbstractInputChannelRecoveredStateHandler.create( inputGates, taskStateSnapshot.getInputRescalingDescriptor(), + filterContext.isCheckpointingDuringRecoveryEnabled(), filteringHandler, filterContext.getMemorySegmentSize())) { read( diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java index d9b7885815b..168901a977b 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannel.java @@ -166,10 +166,7 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan public void onRecoveredStateBuffer(Buffer buffer) { boolean recycleBuffer = true; NetworkActionsLogger.traceRecover( - "InputChannelRecoveredStateHandler#recover", - buffer, - inputGate.getOwningTaskName(), - channelInfo); + "NoSpillingHandler#recover", buffer, inputGate.getOwningTaskName(), channelInfo); try { final boolean wasEmpty; synchronized (receivedBuffers) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java index 9c4aab0bc7a..18f8a863352 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/InputChannelRecoveredStateHandlerTest.java @@ -40,12 +40,13 @@ import static org.apache.flink.runtime.checkpoint.InflightDataRescalingDescripto import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -/** Test of different implementation of {@link InputChannelRecoveredStateHandler}. */ +/** Test of different implementation of {@link AbstractInputChannelRecoveredStateHandler}. */ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandlerTest { private static final int preAllocatedSegments = 3; private NetworkBufferPool networkBufferPool; private SingleInputGate inputGate; - private InputChannelRecoveredStateHandler icsHandler; + // NoSpillingHandler: checkpointingDuringRecoveryEnabled=false, filteringHandler=null + private AbstractInputChannelRecoveredStateHandler icsHandler; private InputChannelInfo channelInfo; @BeforeEach @@ -61,14 +62,14 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .setSegmentProvider(networkBufferPool) .build(); - icsHandler = buildInputChannelStateHandler(inputGate); + icsHandler = buildNoSpillingHandler(inputGate); channelInfo = new InputChannelInfo(0, 0); } - private InputChannelRecoveredStateHandler buildInputChannelStateHandler( + private AbstractInputChannelRecoveredStateHandler buildNoSpillingHandler( SingleInputGate inputGate) { - return new InputChannelRecoveredStateHandler( + return AbstractInputChannelRecoveredStateHandler.create( new InputGate[] {inputGate}, new InflightDataRescalingDescriptor( new InflightDataRescalingDescriptor @@ -82,11 +83,12 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .InflightDataGateOrPartitionRescalingDescriptor .MappingType.IDENTITY) }), + false, null, MemoryManager.DEFAULT_PAGE_SIZE); } - private InputChannelRecoveredStateHandler buildMultiChannelHandler() { + private AbstractInputChannelRecoveredStateHandler buildMultiChannelHandler() { // Setup multi-channel scenario to trigger distribution constraint validation SingleInputGate multiChannelGate = new SingleInputGateBuilder() @@ -95,7 +97,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .setSegmentProvider(networkBufferPool) .build(); - return new InputChannelRecoveredStateHandler( + return AbstractInputChannelRecoveredStateHandler.create( new InputGate[] {multiChannelGate}, new InflightDataRescalingDescriptor( new InflightDataRescalingDescriptor @@ -110,39 +112,42 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .InflightDataGateOrPartitionRescalingDescriptor .MappingType.RESCALING) }), + false, null, MemoryManager.DEFAULT_PAGE_SIZE); } /** Builds a handler in filtering mode (non-null filtering handler, no-op stub). */ - private InputChannelRecoveredStateHandler buildFilteringInputChannelStateHandler() { + private FilteringHandler buildFilteringInputChannelStateHandler() { // Empty GateFilterHandler array: filtering is "enabled" structurally, but no gate-level // filter logic runs. Suitable for exercising getBuffer() routing only. ChannelStateFilteringHandler stubFilteringHandler = new ChannelStateFilteringHandler( new ChannelStateFilteringHandler.GateFilterHandler[0]); - return new InputChannelRecoveredStateHandler( - new InputGate[] {inputGate}, - new InflightDataRescalingDescriptor( - new InflightDataRescalingDescriptor - .InflightDataGateOrPartitionRescalingDescriptor[] { - new InflightDataRescalingDescriptor - .InflightDataGateOrPartitionRescalingDescriptor( - new int[] {1}, - RescaleMappings.identity(1, 1), - new HashSet<>(), - InflightDataRescalingDescriptor - .InflightDataGateOrPartitionRescalingDescriptor - .MappingType.IDENTITY) - }), - stubFilteringHandler, - MemoryManager.DEFAULT_PAGE_SIZE); + return (FilteringHandler) + AbstractInputChannelRecoveredStateHandler.create( + new InputGate[] {inputGate}, + new InflightDataRescalingDescriptor( + new InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor[] { + new InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor( + new int[] {1}, + RescaleMappings.identity(1, 1), + new HashSet<>(), + InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor + .MappingType.IDENTITY) + }), + true, + stubFilteringHandler, + MemoryManager.DEFAULT_PAGE_SIZE); } @Test void testBufferDistributedToMultipleInputChannelsThrowsException() throws Exception { // Test constraint that prevents buffer distribution to multiple channels - try (InputChannelRecoveredStateHandler handler = buildMultiChannelHandler()) { + try (AbstractInputChannelRecoveredStateHandler handler = buildMultiChannelHandler()) { assertThatThrownBy(() -> handler.getBuffer(channelInfo)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining( @@ -187,8 +192,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler @Test void testPreFilterBufferIsolationFromNetworkBufferPool() throws Exception { - try (InputChannelRecoveredStateHandler filteringHandler = - buildFilteringInputChannelStateHandler()) { + try (FilteringHandler filteringHandler = buildFilteringInputChannelStateHandler()) { int availableBefore = networkBufferPool.getNumberOfAvailableMemorySegments(); RecoveredChannelStateHandler.BufferWithContext<Buffer> bufferWithContext = @@ -229,8 +233,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler @Test void testPreFilterSegmentReusedAcrossCalls() throws Exception { - try (InputChannelRecoveredStateHandler filteringHandler = - buildFilteringInputChannelStateHandler()) { + try (FilteringHandler filteringHandler = buildFilteringInputChannelStateHandler()) { // First getBuffer() lazily allocates the segment. RecoveredChannelStateHandler.BufferWithContext<Buffer> first = filteringHandler.getBuffer(channelInfo); @@ -259,8 +262,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler @Test void testGetBufferThrowsWhenPriorBufferNotRecycled() throws Exception { - try (InputChannelRecoveredStateHandler filteringHandler = - buildFilteringInputChannelStateHandler()) { + try (FilteringHandler filteringHandler = buildFilteringInputChannelStateHandler()) { RecoveredChannelStateHandler.BufferWithContext<Buffer> first = filteringHandler.getBuffer(channelInfo); try { @@ -283,8 +285,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler @Test void testPreFilterSegmentFreedOnClose() throws Exception { - InputChannelRecoveredStateHandler filteringHandler = - buildFilteringInputChannelStateHandler(); + FilteringHandler filteringHandler = buildFilteringInputChannelStateHandler(); RecoveredChannelStateHandler.BufferWithContext<Buffer> bufferWithContext = filteringHandler.getBuffer(channelInfo); bufferWithContext.context.recycleBuffer(); diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java index 01f7c43920a..db302d23f73 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandlerTest.java @@ -22,7 +22,7 @@ import org.junit.jupiter.api.Test; /** * Base class which contains all tests which should be implemented for every implementation of - * {@link InputChannelRecoveredStateHandler}. + * {@link AbstractInputChannelRecoveredStateHandler}. */ abstract class RecoveredChannelStateHandlerTest {
