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 af146b61707d5938481ec50acfb589d26f3480fe Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 02:29:17 2026 +0200 [FLINK-40080] Switch flag-on recovery to fetch->drain; retire the in-memory backend One semantic unit (the backend swap), atomically: - AbstractInputChannelRecoveredStateHandler.create(...): flag-on now returns SpillingWithFilteringHandler / SpillingNoFilteringHandler; delete the v1 FilteringHandler and the duplicated v1 delivery path (ChannelStateFilteringHandler's buffer-delivering filterAndRewrite overload, BufferSupplier, and the chunking machinery). - StreamTask.recoverChannelsWithCheckpointing: NOT_READY -> fetchChannelState on the channelIOExecutor (now uses the Optional<FetchedChannelState> return) -> requestPartitions( state.isPresent()) on the mailbox -> buildDrainer -> install the drainer as the live recovery-checkpoint trigger -> drain() on the channelIOExecutor -> completeAll(gates' stateConsumedFutures) -> NO_OP. The empty-input-gates synchronous short-circuit is retained. Delete the in-memory trigger and the transitional finishRecoveredBufferDelivery helper (the drainer owns sentinel appending at the end of its drain). - RecoveredInputChannel.toInputChannel(true): flag-on queues are now always empty (state goes to disk, not to the queues) -> delete the conversion-time push loop; unify on checkState(receivedBuffers.isEmpty()). --- .../channel/ChannelStateFilteringHandler.java | 225 ------------- .../channel/RecoveredChannelStateHandler.java | 193 +---------- .../partition/consumer/RecoveredInputChannel.java | 42 +-- .../InputChannelRecoveredStateHandlerTest.java | 365 ++++++++++++++++----- .../consumer/LocalRecoveredInputChannelTest.java | 2 +- .../consumer/RecoveredInputChannelTest.java | 121 +++++-- .../consumer/RemoteRecoveredInputChannelTest.java | 2 +- 7 files changed, 401 insertions(+), 549 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java index ef3db27bbff..0b6976068d7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateFilteringHandler.java @@ -133,48 +133,6 @@ public class ChannelStateFilteringHandler implements Closeable { oldSubtaskIndex, oldChannelIndex, sourceBuffer, outputSerializer); } - /** - * FLINK-38544 transitional: removed when the spilling backend lands (dies together with the - * in-memory {@code FilteringHandler}, its only caller). - * - * <p>Filters a recovered buffer from the specified virtual channel, returning new buffers - * containing only the records that belong to the current subtask. - * - * <p>One source buffer may produce 0 to N result buffers: 0 if all records are filtered out, - * and potentially more than 1 when a spanning record completes in this buffer. The deserializer - * caches partial record data from previous buffers, so the output may contain data that was not - * in the current source buffer, causing the total output size to exceed one buffer capacity. - * This can happen with any spanning record regardless of its size. - * - * @return filtered buffers, possibly empty if all records were filtered out. - */ - public List<Buffer> filterAndRewrite( - int gateIndex, - int oldSubtaskIndex, - int oldChannelIndex, - Buffer sourceBuffer, - BufferSupplier bufferSupplier) - throws IOException, InterruptedException { - - if (gateIndex < 0 || gateIndex >= gateHandlers.length) { - throw new IllegalStateException( - "Invalid gateIndex: " - + gateIndex - + ", number of gates: " - + gateHandlers.length); - } - - GateFilterHandler<?> gateHandler = gateHandlers[gateIndex]; - if (gateHandler == null) { - throw new IllegalStateException( - "No handler for gateIndex " - + gateIndex - + ". This gate is not a network input and should not have recovered buffers."); - } - return gateHandler.filterAndRewrite( - oldSubtaskIndex, oldChannelIndex, sourceBuffer, bufferSupplier); - } - /** Returns {@code true} if any virtual channel has a partial (spanning) record pending. */ public boolean hasPartialData() { for (GateFilterHandler<?> handler : gateHandlers) { @@ -287,15 +245,6 @@ public class ChannelStateFilteringHandler implements Closeable { // Inner classes // ------------------------------------------------------------------------------------------- - /** - * FLINK-38544 transitional: removed when the spilling backend lands. Provides buffers for - * re-serializing filtered records on the in-memory delivery path. Implementations may block. - */ - @FunctionalInterface - public interface BufferSupplier { - Buffer requestBufferBlocking() throws IOException, InterruptedException; - } - /** * Handles record filtering for a single input gate. Each gate has its own serializer and set of * virtual channels, allowing different gates to handle different record types independently. @@ -306,18 +255,12 @@ public class ChannelStateFilteringHandler implements Closeable { private final StreamElementSerializer<T> serializer; private final DeserializationDelegate<StreamElement> deserializationDelegate; - // FLINK-38544 transitional: removed when the spilling backend lands (used only by the - // in-memory buffer-delivering filterAndRewrite overload below). - private final DataOutputSerializer outputSerializer; - private final byte[] lengthBuffer = new byte[4]; - GateFilterHandler( Map<SubtaskConnectionDescriptor, VirtualChannel<T>> virtualChannels, StreamElementSerializer<T> serializer) { this.virtualChannels = checkNotNull(virtualChannels); this.serializer = checkNotNull(serializer); this.deserializationDelegate = new NonReusingDeserializationDelegate<>(serializer); - this.outputSerializer = new DataOutputSerializer(128); } /** @@ -380,174 +323,6 @@ public class ChannelStateFilteringHandler implements Closeable { outputSerializer.writeIntUnsafe(recordLength, startPos); } - // ----------------------------------------------------------------------------------- - // FLINK-38544 transitional: everything below until hasPartialData() is the in-memory - // buffer-delivering filter path, kept alive for the v1 FilteringHandler; removed when - // the spilling backend lands. - // ----------------------------------------------------------------------------------- - - /** - * Deserializes records from {@code sourceBuffer}, applies the virtual channel's record - * filter, and immediately re-serializes each surviving record into output buffers. - */ - List<Buffer> filterAndRewrite( - int oldSubtaskIndex, - int oldChannelIndex, - Buffer sourceBuffer, - BufferSupplier bufferSupplier) - throws IOException, InterruptedException { - - boolean sourceBufferOwnershipTransferred = false; - List<Buffer> resultBuffers = new ArrayList<>(); - Buffer currentBuffer = null; - try { - SubtaskConnectionDescriptor key = - new SubtaskConnectionDescriptor(oldSubtaskIndex, oldChannelIndex); - VirtualChannel<T> vc = virtualChannels.get(key); - if (vc == null) { - throw new IllegalStateException( - "No VirtualChannel found for key: " - + key - + "; known channels are " - + virtualChannels.keySet()); - } - - vc.setNextBuffer(sourceBuffer); - sourceBufferOwnershipTransferred = true; - - while (true) { - DeserializationResult result = vc.getNextRecord(deserializationDelegate); - if (result.isFullRecord()) { - if (currentBuffer == null) { - currentBuffer = bufferSupplier.requestBufferBlocking(); - } - currentBuffer = - serializeElement( - deserializationDelegate.getInstance(), - currentBuffer, - resultBuffers, - bufferSupplier); - } - if (result.isBufferConsumed()) { - break; - } - } - - if (currentBuffer != null) { - if (currentBuffer.readableBytes() > 0) { - resultBuffers.add(currentBuffer); - } else { - currentBuffer.recycleBuffer(); - } - currentBuffer = null; - } - - return resultBuffers; - } catch (Throwable t) { - if (!sourceBufferOwnershipTransferred) { - sourceBuffer.recycleBuffer(); - } - // Avoid double-recycle: currentBuffer may already be the last element in - // resultBuffers if serializeElement added it before the exception. - if (currentBuffer != null - && (resultBuffers.isEmpty() - || resultBuffers.get(resultBuffers.size() - 1) != currentBuffer)) { - currentBuffer.recycleBuffer(); - } - for (Buffer buf : resultBuffers) { - buf.recycleBuffer(); - } - resultBuffers.clear(); - throw t; - } - } - - /** - * Serializes a single stream element into the current buffer using the length-prefixed - * format (4-byte big-endian length + record bytes) expected by Flink's record - * deserializers. Spills into new buffers from {@code bufferSupplier} when needed. - * - * @return the buffer to continue writing into (may differ from the input buffer). - */ - private Buffer serializeElement( - StreamElement element, - Buffer currentBuffer, - List<Buffer> resultBuffers, - BufferSupplier bufferSupplier) - throws IOException, InterruptedException { - outputSerializer.clear(); - serializer.serialize(element, outputSerializer); - int recordLength = outputSerializer.length(); - - writeLengthToBuffer(recordLength); - currentBuffer = - writeDataToBuffer( - lengthBuffer, 0, 4, currentBuffer, resultBuffers, bufferSupplier); - - byte[] serializedData = outputSerializer.getSharedBuffer(); - currentBuffer = - writeDataToBuffer( - serializedData, - 0, - recordLength, - currentBuffer, - resultBuffers, - bufferSupplier); - return currentBuffer; - } - - private void writeLengthToBuffer(int length) { - lengthBuffer[0] = (byte) (length >> 24); - lengthBuffer[1] = (byte) (length >> 16); - lengthBuffer[2] = (byte) (length >> 8); - lengthBuffer[3] = (byte) length; - } - - /** - * Writes data to the current buffer, spilling into new buffers from {@code bufferSupplier} - * when the current one is full. - * - * @return the buffer to continue writing into (may differ from the input buffer). - */ - private Buffer writeDataToBuffer( - byte[] data, - int dataOffset, - int dataLength, - Buffer currentBuffer, - List<Buffer> resultBuffers, - BufferSupplier bufferSupplier) - throws IOException, InterruptedException { - int offset = dataOffset; - int remaining = dataLength; - - while (remaining > 0) { - int writableBytes = currentBuffer.getMaxCapacity() - currentBuffer.getSize(); - - if (writableBytes == 0) { - // Buffer is full, transfer ownership to resultBuffers - resultBuffers.add(currentBuffer); - currentBuffer = bufferSupplier.requestBufferBlocking(); - writableBytes = currentBuffer.getMaxCapacity(); - } - - int bytesToWrite = Math.min(remaining, writableBytes); - currentBuffer - .getMemorySegment() - .put( - currentBuffer.getMemorySegmentOffset() + currentBuffer.getSize(), - data, - offset, - bytesToWrite); - currentBuffer.setSize(currentBuffer.getSize() + bytesToWrite); - - offset += bytesToWrite; - remaining -= bytesToWrite; - } - return currentBuffer; - } - - // ------------------------ end of FLINK-38544 transitional block ----------------------- - boolean hasPartialData() { return virtualChannels.values().stream().anyMatch(VirtualChannel::hasPartialData); } 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 4a5c005c303..23960227843 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 @@ -100,20 +100,10 @@ abstract class AbstractInputChannelRecoveredStateHandler final Map<InputChannelInfo, RecoveredInputChannel> rescaledChannels = new HashMap<>(); final Map<Integer, RescaleMappings> oldToNewMappings = new HashMap<>(); - /** - * FLINK-38544 transitional: removed when the spilling backend lands. Gates the unbounded - * heap-buffer fallback in {@link RecoveredInputChannel#requestBufferBlocking(boolean)}; read - * from the job configuration at the call site now that the gate-level recovery flags are gone. - */ - final boolean checkpointingDuringRecoveryEnabled; - AbstractInputChannelRecoveredStateHandler( - InputGate[] inputGates, - InflightDataRescalingDescriptor channelMapping, - boolean checkpointingDuringRecoveryEnabled) { + InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) { this.inputGates = inputGates; this.channelMapping = channelMapping; - this.checkpointingDuringRecoveryEnabled = checkpointingDuringRecoveryEnabled; } /** @@ -122,8 +112,9 @@ abstract class AbstractInputChannelRecoveredStateHandler * * <ul> * <li>{@code false} → {@link NoSpillingHandler} - * <li>{@code true} and {@code filteringHandler == null} → {@link NoSpillingHandler} - * <li>{@code true} and {@code filteringHandler != null} → {@link FilteringHandler} + * <li>{@code true} and {@code filteringHandler == null} → {@link SpillingNoFilteringHandler} + * <li>{@code true} and {@code filteringHandler != null} → {@link + * SpillingWithFilteringHandler} * </ul> */ static AbstractInputChannelRecoveredStateHandler create( @@ -134,15 +125,17 @@ abstract class AbstractInputChannelRecoveredStateHandler int memorySegmentSize, String[] spillTmpDirectories) { if (!checkpointingDuringRecoveryEnabled) { - return new NoSpillingHandler(inputGates, channelMapping, false); + return new NoSpillingHandler(inputGates, channelMapping); } - // FLINK-38544 transitional: the flag-on path still uses the in-memory handlers until the - // spilling backend lands; spillTmpDirectories is unused until then. if (filteringHandler == null) { - return new NoSpillingHandler(inputGates, channelMapping, true); + return new SpillingNoFilteringHandler(inputGates, channelMapping, spillTmpDirectories); } - return new FilteringHandler( - inputGates, channelMapping, filteringHandler, memorySegmentSize); + return new SpillingWithFilteringHandler( + inputGates, + channelMapping, + filteringHandler, + memorySegmentSize, + spillTmpDirectories); } /** Default buffer allocation from the network buffer pool, used by non-filtering modes. */ @@ -150,7 +143,9 @@ abstract class AbstractInputChannelRecoveredStateHandler public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo) throws IOException, InterruptedException { RecoveredInputChannel channel = getMappedChannels(channelInfo); - Buffer buffer = channel.requestBufferBlocking(checkpointingDuringRecoveryEnabled); + // FLINK-38544 transitional: 'false' bypasses the unbounded heap fallback; the parameter + // goes away together with the fallback, which disk spilling supersedes. + Buffer buffer = channel.requestBufferBlocking(false); return new BufferWithContext<>(wrap(buffer), buffer); } @@ -204,11 +199,8 @@ abstract class AbstractInputChannelRecoveredStateHandler */ class NoSpillingHandler extends AbstractInputChannelRecoveredStateHandler { - NoSpillingHandler( - InputGate[] inputGates, - InflightDataRescalingDescriptor channelMapping, - boolean checkpointingDuringRecoveryEnabled) { - super(inputGates, channelMapping, checkpointingDuringRecoveryEnabled); + NoSpillingHandler(InputGate[] inputGates, InflightDataRescalingDescriptor channelMapping) { + super(inputGates, channelMapping); } @Override @@ -315,9 +307,7 @@ abstract class AbstractSpillingHandler extends AbstractInputChannelRecoveredStat String[] spillTmpDirectories, long maxFileSizeBytes, int maxSegmentSizeBytes) { - // FLINK-38544 transitional: the base's third ctor arg is removed when the spilling backend - // lands (spilling always implies checkpointing-during-recovery enabled). - super(inputGates, channelMapping, true); + super(inputGates, channelMapping); checkArgument( checkNotNull(spillTmpDirectories).length > 0, "spillTmpDirectories must not be empty"); @@ -623,153 +613,6 @@ class SpillingWithFilteringHandler extends AbstractSpillingHandler { } } -/** - * FLINK-38544 transitional: removed when the spilling backend lands (the factory then selects - * {@link SpillingWithFilteringHandler} instead). - * - * <p>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 #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}. - */ - @Nullable private MemorySegment preFilterSegment; - - /** - * Tracks whether {@link #preFilterSegment} is currently wrapped by a live {@link Buffer} that - * has not yet been recycled. Flipped to {@code true} when a new buffer is issued, and flipped - * back to {@code false} by the custom {@link BufferRecycler} when the buffer is recycled. - */ - private boolean preFilterBufferInUse; - - FilteringHandler( - InputGate[] inputGates, - InflightDataRescalingDescriptor channelMapping, - ChannelStateFilteringHandler filteringHandler, - int memorySegmentSize) { - super(inputGates, channelMapping, true); - this.filteringHandler = filteringHandler; - checkArgument( - memorySegmentSize > 0, "memorySegmentSize must be positive: %s", memorySegmentSize); - this.memorySegmentSize = memorySegmentSize; - } - - /** - * 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 #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 - * asserts the invariant before issuing a buffer: if a previously issued buffer has not yet been - * recycled, it throws {@link IllegalStateException} so any future regression fails loudly - * instead of silently corrupting memory. - */ - @Override - public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo) { - checkState( - !preFilterBufferInUse, - "Previous pre-filter buffer has not been recycled. This violates the " - + "one-buffer-at-a-time invariant of pre-filter buffers."); - - if (preFilterSegment == null) { - preFilterSegment = MemorySegmentFactory.allocateUnpooledSegment(memorySegmentSize); - } - preFilterBufferInUse = true; - - // The recycler keeps the segment alive for reuse; only flips the in-use flag. - BufferRecycler recycler = segment -> preFilterBufferInUse = false; - Buffer buffer = new NetworkBuffer(preFilterSegment, recycler); - return new BufferWithContext<>(wrap(buffer), buffer); - } - - @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); - recoverWithFiltering(channel, channelInfo, oldSubtaskIndex, buffer.retainBuffer()); - } - } finally { - buffer.recycleBuffer(); - } - } - - private void recoverWithFiltering( - RecoveredInputChannel channel, - InputChannelInfo channelInfo, - int oldSubtaskIndex, - Buffer retainedBuffer) - throws IOException, InterruptedException { - checkState(filteringHandler != null, "filtering handler not set."); - List<Buffer> filteredBuffers = - filteringHandler.filterAndRewrite( - channelInfo.getGateIdx(), - oldSubtaskIndex, - channelInfo.getInputChannelIdx(), - retainedBuffer, - () -> channel.requestBufferBlocking(checkpointingDuringRecoveryEnabled)); - - int i = 0; - try { - for (; i < filteredBuffers.size(); i++) { - channel.onRecoveredStateBuffer(filteredBuffers.get(i)); - } - } catch (Throwable t) { - // Start at i + 1: onRecoveredStateBuffer() takes over the buffer before anything can - // fail, so recycling the buffer at index i again would corrupt its reference count. - for (int j = i + 1; j < filteredBuffers.size(); j++) { - filteredBuffers.get(j).recycleBuffer(); - } - throw t; - } - } - - @VisibleForTesting - boolean isPreFilterBufferInUse() { - return preFilterBufferInUse; - } - - @VisibleForTesting - @Nullable - MemorySegment getPreFilterSegmentForTesting() { - return preFilterSegment; - } - - @Override - void closeInternal() throws IOException { - if (preFilterSegment != null) { - preFilterSegment.free(); - preFilterSegment = null; - preFilterBufferInUse = false; - } - } -} - class ResultSubpartitionRecoveredStateHandler implements RecoveredChannelStateHandler<ResultSubpartitionInfo, BufferBuilder> { 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 31922213406..2c6355e8953 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 @@ -109,9 +109,7 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan } public final InputChannel toInputChannel(boolean needsRecovery) throws IOException { - if (needsRecovery) { - return toInputChannelInRecovery(); - } + // With checkpointing-during-recovery, data is spilled instead of queued here. synchronized (receivedBuffers) { Preconditions.checkState(receivedBuffers.isEmpty(), "Received buffer should be empty."); } @@ -122,44 +120,6 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan return inputChannel; } - /** - * FLINK-38544 transitional: removed when the spilling backend lands. Creates the physical - * channel in recovery state and synchronously hands every queued recovered data buffer over - * through the push interface. The legacy {@link EndOfInputChannelStateEvent} in the queue is - * dropped in translation; the {@link EndOfFetchedChannelStateEvent} sentinel takes its place - * but is deliberately NOT appended here: the StreamTask recovery chain appends it via {@link - * RecoverableInputChannel#finishRecoveredBufferDelivery()} on the channel IO executor after - * partitions have been requested. That call waits for upstream readiness, which (a) cannot - * happen on the mailbox thread that is still converting channels (partitions are requested only - * after conversion) and (b) must happen before the sentinel becomes consumable -- otherwise the - * consume path could flip the channel out of recovery and poll it before its upstream - * connection exists. - */ - private InputChannel toInputChannelInRecovery() throws IOException { - final Buffer[] remainingBuffers; - synchronized (receivedBuffers) { - remainingBuffers = receivedBuffers.toArray(new Buffer[0]); - receivedBuffers.clear(); - } - - final InputChannel inputChannel = toInputChannelInternal(true); - inputChannel.setup(); - final RecoverableInputChannel recoverableChannel = (RecoverableInputChannel) inputChannel; - for (int i = 0; i < remainingBuffers.length; i++) { - final Buffer buffer = remainingBuffers[i]; - if (isEndOfInputChannelStateEvent(buffer)) { - Preconditions.checkState( - i == remainingBuffers.length - 1, - "EndOfInputChannelStateEvent must be the last recovered buffer."); - buffer.recycleBuffer(); - } else { - recoverableChannel.onRecoveredStateBuffer(buffer); - } - } - inputChannel.checkpointStopped(lastStoppedCheckpointId); - return inputChannel; - } - @Override public void checkpointStopped(long checkpointId) { this.lastStoppedCheckpointId = checkpointId; 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 e398e41562a..61fe42a6a4c 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 @@ -18,31 +18,46 @@ package org.apache.flink.runtime.checkpoint.channel; +import org.apache.flink.api.common.typeutils.base.LongSerializer; +import org.apache.flink.core.memory.DataOutputSerializer; import org.apache.flink.core.memory.MemorySegment; +import org.apache.flink.core.memory.MemorySegmentFactory; import org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptor; import org.apache.flink.runtime.checkpoint.RescaleMappings; +import org.apache.flink.runtime.io.network.api.SubtaskConnectionDescriptor; +import org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer; +import org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer; import org.apache.flink.runtime.io.network.buffer.Buffer; +import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler; import org.apache.flink.runtime.io.network.buffer.NetworkBuffer; import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool; -import org.apache.flink.runtime.io.network.partition.consumer.FailingRecoveredInputChannel; +import org.apache.flink.runtime.io.network.partition.consumer.InputChannel; import org.apache.flink.runtime.io.network.partition.consumer.InputChannelBuilder; import org.apache.flink.runtime.io.network.partition.consumer.InputGate; +import org.apache.flink.runtime.io.network.partition.consumer.RecoveredInputChannel; import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate; import org.apache.flink.runtime.io.network.partition.consumer.SingleInputGateBuilder; import org.apache.flink.runtime.memory.MemoryManager; +import org.apache.flink.runtime.plugable.DeserializationDelegate; +import org.apache.flink.streaming.runtime.io.recovery.RecordFilter; +import org.apache.flink.streaming.runtime.io.recovery.VirtualChannel; +import org.apache.flink.streaming.runtime.streamrecord.StreamElement; +import org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; import java.nio.file.Path; -import java.util.Arrays; +import java.util.HashMap; import java.util.HashSet; -import java.util.List; +import java.util.Map; +import java.util.Optional; import static org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptorUtil.mappings; import static org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptorUtil.to; -import static org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils.buildSomeBuffer; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -134,35 +149,7 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler ChannelStateFilteringHandler stubFilteringHandler = new ChannelStateFilteringHandler( new ChannelStateFilteringHandler.GateFilterHandler[0]); - // FLINK-38544 transitional: constructed directly because the factory still routes the - // flag-on filtering case to the in-memory FilteringHandler; goes back through - // AbstractInputChannelRecoveredStateHandler.create(...) when the spilling backend lands. - return new SpillingWithFilteringHandler( - 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, - new String[] {tmpDir.toAbsolutePath().toString()}); - } - - /** - * Builds the in-memory filtering handler through the factory: the flag-on filtering case still - * routes there until the spilling backend lands. - */ - private FilteringHandler buildFilteringInputChannelStateHandler( - SingleInputGate inputGate, ChannelStateFilteringHandler stubFilteringHandler) { - return (FilteringHandler) + return (SpillingWithFilteringHandler) AbstractInputChannelRecoveredStateHandler.create( new InputGate[] {inputGate}, new InflightDataRescalingDescriptor( @@ -180,15 +167,12 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler true, stubFilteringHandler, MemoryManager.DEFAULT_PAGE_SIZE, - null); + new String[] {tmpDir.toAbsolutePath().toString()}); } private AbstractInputChannelRecoveredStateHandler buildSpillingNoFilteringHandler( String[] spillTmpDirectories) { - // FLINK-38544 transitional: constructed directly because the factory still routes the - // flag-on no-filtering case to the in-memory NoSpillingHandler; goes back through - // AbstractInputChannelRecoveredStateHandler.create(...) when the spilling backend lands. - return new SpillingNoFilteringHandler( + return AbstractInputChannelRecoveredStateHandler.create( new InputGate[] {inputGate}, new InflightDataRescalingDescriptor( new InflightDataRescalingDescriptor @@ -202,6 +186,9 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .InflightDataGateOrPartitionRescalingDescriptor .MappingType.IDENTITY) }), + true, + null, + MemoryManager.DEFAULT_PAGE_SIZE, spillTmpDirectories); } @@ -347,49 +334,6 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler } } - @Test - void testFilteredBuffersRecycledOnceWhenDeliveryFails() throws Exception { - List<Buffer> filteredBuffers = - Arrays.asList(buildSomeBuffer(), buildSomeBuffer(), buildSomeBuffer()); - ChannelStateFilteringHandler filteringHandler = - new ChannelStateFilteringHandler( - new ChannelStateFilteringHandler.GateFilterHandler[0]) { - @Override - public List<Buffer> filterAndRewrite( - int gateIndex, - int oldSubtaskIndex, - int oldChannelIndex, - Buffer sourceBuffer, - BufferSupplier bufferSupplier) { - sourceBuffer.recycleBuffer(); - return filteredBuffers; - } - }; - // The gate fails while taking over the first filtered buffer. - SingleInputGate failingGate = - new SingleInputGateBuilder() - .setChannelFactory( - (builder, gate) -> new FailingRecoveredInputChannel(gate, 0)) - .setSegmentProvider(networkBufferPool) - .build(); - - try (FilteringHandler handler = - buildFilteringInputChannelStateHandler(failingGate, filteringHandler)) { - RecoveredChannelStateHandler.BufferWithContext<Buffer> bufferWithContext = - handler.getBuffer(channelInfo); - bufferWithContext.context.setSize(1); - - assertThatThrownBy(() -> handler.recover(channelInfo, 0, bufferWithContext)) - .isInstanceOf(IllegalStateException.class) - .hasMessage("Delivery failed on purpose."); - } - - // The first buffer is owned by the channel, the undelivered ones are recycled exactly once. - assertThat(filteredBuffers.get(0).isRecycled()).isFalse(); - assertThat(filteredBuffers.get(1).isRecycled()).isTrue(); - assertThat(filteredBuffers.get(2).isRecycled()).isTrue(); - } - @Test void testPreFilterSegmentFreedOnClose() throws Exception { SpillingWithFilteringHandler filteringHandler = buildFilteringInputChannelStateHandler(); @@ -415,4 +359,263 @@ class InputChannelRecoveredStateHandlerTest extends RecoveredChannelStateHandler .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("spillTmpDirectories must not be empty"); } + + // ------------------------------------------------------------------------------------------- + // Filter-on / filter-off routing. + // + // These use a dedicated large-pool fixture (a pass-through filter on filter-on must not + // deadlock on a bounded pool), independent of the small-pool fixture in setUp(). + // ------------------------------------------------------------------------------------------- + + @Test + void testFilterOnRoutesOutputToChannelState() throws Exception { + try (RoutingFixture fx = newRoutingFixture()) { + ChannelStateFilteringHandler filteringHandler = fx.newPassThroughFilteringHandler(); + try (ChannelStateFilteringHandler ignored = filteringHandler) { + SpillingWithFilteringHandler handler = fx.newFilterOnHandler(filteringHandler); + fx.invokeRecoverWithRecords(handler, 1L, 2L, 3L); + + // Surviving records accumulate in the in-memory segment serializer; they are only + // sealed and flushed to a spill file on channel switch or close. With a single + // channel and no switch, close() is what seals the segment, so the assertion must + // follow it. + handler.close(); + + assertThat(handler.peekSpillFilesForTesting()) + .as("filter-on path must spill the surviving records to a file") + .isNotEmpty(); + } + } + } + + @Test + void testFilterOnAccumulatorBuffersComeFromHeapNotPool() throws Exception { + try (RoutingFixture fx = newRoutingFixture()) { + // The accumulator's prefilter + postfilter buffers are unpooled heap segments owned by + // the handler — invoking filter recovery must NOT consume any network buffer pool + // segments for the accumulator path. + ChannelStateFilteringHandler filteringHandler = fx.newPassThroughFilteringHandler(); + SpillingWithFilteringHandler handler = fx.newFilterOnHandler(filteringHandler); + try (ChannelStateFilteringHandler ignored = filteringHandler) { + int availableBeforeRecover = fx.pool.getNumberOfAvailableMemorySegments(); + + fx.invokeRecoverWithRecords(handler, 1L, 2L, 3L); + + assertThat(fx.pool.getNumberOfAvailableMemorySegments()) + .as("filter accumulator buffers must not be sourced from the network pool") + .isEqualTo(availableBeforeRecover); + + handler.close(); + fx.gate.close(); + assertThat(fx.pool.getNumberOfAvailableMemorySegments()) + .as("pool count after close must match pre-recover (filter took nothing)") + .isEqualTo(availableBeforeRecover); + } + } + } + + @Test + void testFilterOnDoesNotInvokeChannelOnRecoveredStateBuffer() throws Exception { + try (RoutingFixture fx = newRoutingFixture()) { + ChannelStateFilteringHandler filteringHandler = fx.newPassThroughFilteringHandler(); + try (ChannelStateFilteringHandler ignored = filteringHandler; + SpillingWithFilteringHandler handler = + fx.newFilterOnHandler(filteringHandler)) { + fx.invokeRecoverWithRecords(handler, 1L, 2L, 3L); + + assertThat(fx.countQueuedRecoveredBuffers()) + .as("filter-on must not enqueue buffers into the channel during recovery") + .isEqualTo(0); + } + } + } + + @Test + void testFilterOffMaintainsMasterBehavior() throws Exception { + try (RoutingFixture fx = newRoutingFixture(); + NoSpillingHandler handler = fx.newFilterOffHandler()) { + fx.invokeRecoverWithRawBytes(handler, new byte[] {1, 2, 3, 4}); + + // Filter-off path enqueues the SubtaskConnectionDescriptor event plus the data buffer + // directly into the channel's recoveredBuffers. + assertThat(fx.countQueuedRecoveredBuffers()) + .as("filter-off must enqueue the descriptor + data buffer into the channel") + .isGreaterThanOrEqualTo(2); + } + } + + private RoutingFixture newRoutingFixture() { + return new RoutingFixture(); + } + + /** + * Self-contained fixture for the filter routing tests: a large network buffer pool (so a + * pass-through filter-on path does not deadlock on a bounded pool) with its own recovered input + * gate and channel. + */ + private static final class RoutingFixture implements AutoCloseable { + private final NetworkBufferPool pool = + new NetworkBufferPool(64, MemoryManager.DEFAULT_PAGE_SIZE); + private final SingleInputGate gate = + new SingleInputGateBuilder() + .setChannelFactory(InputChannelBuilder::buildLocalRecoveredChannel) + .setSegmentProvider(pool) + .build(); + private final InputChannelInfo channelInfo = new InputChannelInfo(0, 0); + + private final Path spillDir; + + RoutingFixture() { + try { + spillDir = java.nio.file.Files.createTempDirectory("filter-routing-"); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + SpillingWithFilteringHandler newFilterOnHandler( + ChannelStateFilteringHandler filteringHandler) { + return (SpillingWithFilteringHandler) + AbstractInputChannelRecoveredStateHandler.create( + new InputGate[] {gate}, + identityRescalingForOneGate(), + true, + filteringHandler, + MemoryManager.DEFAULT_PAGE_SIZE, + new String[] {spillDir.toString()}); + } + + NoSpillingHandler newFilterOffHandler() { + return (NoSpillingHandler) + AbstractInputChannelRecoveredStateHandler.create( + new InputGate[] {gate}, + identityRescalingForOneGate(), + false, + null, + MemoryManager.DEFAULT_PAGE_SIZE, + null); + } + + ChannelStateFilteringHandler newPassThroughFilteringHandler() { + StreamElementSerializer<Long> serializer = + new StreamElementSerializer<>(LongSerializer.INSTANCE); + RecordDeserializer<DeserializationDelegate<StreamElement>> deserializer = + new SpillingAdaptiveSpanningRecordDeserializer<>( + new String[] {System.getProperty("java.io.tmpdir")}); + VirtualChannel<Long> vc = new VirtualChannel<>(deserializer, RecordFilter.acceptAll()); + Map<SubtaskConnectionDescriptor, VirtualChannel<Long>> channels = new HashMap<>(); + // The handler invokes filterAndRewrite with oldSubtaskIndex=1 — keep the key aligned. + channels.put(new SubtaskConnectionDescriptor(1, channelInfo.getInputChannelIdx()), vc); + + ChannelStateFilteringHandler.GateFilterHandler<Long> gateHandler = + new ChannelStateFilteringHandler.GateFilterHandler<>(channels, serializer); + return new ChannelStateFilteringHandler( + new ChannelStateFilteringHandler.GateFilterHandler<?>[] {gateHandler}); + } + + void invokeRecoverWithRecords( + AbstractInputChannelRecoveredStateHandler handler, Long... values) + throws Exception { + invokeRecoverWithBuffer(handler, createRecordBuffer(values)); + } + + void invokeRecoverWithRawBytes( + AbstractInputChannelRecoveredStateHandler handler, byte[] data) throws Exception { + MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(data.length); + seg.put(0, data); + NetworkBuffer source = new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE); + source.setSize(data.length); + invokeRecoverWithBuffer(handler, source); + } + + /** + * Mirrors the chunkReader's getBuffer + recover sequence: the handler-issued buffer is + * filled with the source data, then handed back to recover. + */ + private void invokeRecoverWithBuffer( + AbstractInputChannelRecoveredStateHandler handler, Buffer source) throws Exception { + RecoveredChannelStateHandler.BufferWithContext<Buffer> bwc = + handler.getBuffer(channelInfo); + try { + Buffer dest = bwc.context; + int len = source.readableBytes(); + source.getMemorySegment() + .copyTo( + source.getMemorySegmentOffset(), + dest.getMemorySegment(), + dest.getMemorySegmentOffset(), + len); + dest.setSize(len); + } finally { + source.recycleBuffer(); + } + // oldSubtaskIndex=1 matches the pass-through filter's virtual channel key. The + // filter-off path ignores this argument's mapping (it only flows into a + // SubtaskConnectionDescriptor). + handler.recover(channelInfo, 1, bwc); + } + + private Buffer createRecordBuffer(Long... values) throws IOException { + StreamElementSerializer<Long> serializer = + new StreamElementSerializer<>(LongSerializer.INSTANCE); + DataOutputSerializer output = new DataOutputSerializer(256); + for (Long v : values) { + DataOutputSerializer rec = new DataOutputSerializer(64); + serializer.serialize(new StreamRecord<>(v), rec); + int recordLength = rec.length(); + output.writeInt(recordLength); + output.write(rec.getSharedBuffer(), 0, recordLength); + } + byte[] data = output.getCopyOfBuffer(); + MemorySegment segment = + MemorySegmentFactory.allocateUnpooledSegment(MemoryManager.DEFAULT_PAGE_SIZE); + segment.put(0, data, 0, data.length); + NetworkBuffer buf = new NetworkBuffer(segment, FreeingBufferRecycler.INSTANCE); + buf.setSize(data.length); + return buf; + } + + /** + * Counts the buffers currently queued in the only recovered input channel by draining via + * the public {@code getNextBuffer()} entry point. After this returns the channel queue is + * empty by definition. + */ + int countQueuedRecoveredBuffers() throws IOException { + RecoveredInputChannel ch = (RecoveredInputChannel) gate.getChannel(0); + int count = 0; + while (true) { + Optional<InputChannel.BufferAndAvailability> next = ch.getNextBuffer(); + if (!next.isPresent()) { + break; + } + count++; + next.get().buffer().recycleBuffer(); + if (count > 1000) { + throw new IllegalStateException("Unexpected unbounded queue contents"); + } + } + return count; + } + + private static InflightDataRescalingDescriptor identityRescalingForOneGate() { + return new InflightDataRescalingDescriptor( + new InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor[] { + new InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor( + new int[] {1}, + RescaleMappings.identity(1, 1), + new HashSet<>(), + InflightDataRescalingDescriptor + .InflightDataGateOrPartitionRescalingDescriptor.MappingType + .IDENTITY) + }); + } + + @Override + public void close() throws Exception { + gate.close(); + pool.destroy(); + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java index fdacf6ec1ee..005c86d3ed0 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalRecoveredInputChannelTest.java @@ -41,7 +41,7 @@ class LocalRecoveredInputChannelTest { try { recoveredChannel.finishReadRecoveredState(); - assertThatThrownBy(() -> recoveredChannel.toInputChannel(false)) + assertThatThrownBy(() -> recoveredChannel.toInputChannel(true)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Received buffer should be empty"); } finally { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java index 15f1b8dc925..c4fee251a5f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RecoveredInputChannelTest.java @@ -24,12 +24,18 @@ import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.io.network.api.CheckpointBarrier; import org.apache.flink.runtime.io.network.buffer.Buffer; import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils; +import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool; import org.apache.flink.runtime.io.network.partition.ResultPartitionID; import org.apache.flink.runtime.io.network.partition.ResultSubpartitionIndexSet; +import org.apache.flink.runtime.memory.MemoryManager; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.apache.flink.runtime.checkpoint.CheckpointOptions.unaligned; import static org.apache.flink.runtime.state.CheckpointStorageLocationReference.getDefault; @@ -39,6 +45,16 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link RecoveredInputChannel}. */ class RecoveredInputChannelTest { + private NetworkBufferPool pool; + + @AfterEach + void tearDown() { + if (pool != null) { + pool.destroy(); + pool = null; + } + } + @Test void testRequestPartitionsImpossible() { assertThatThrownBy(() -> buildChannel(false).requestSubpartitions()) @@ -71,7 +87,7 @@ class RecoveredInputChannelTest { assertThat(channel.getStateConsumedFuture()).isNotDone(); // Conversion fails because the sentinel is still queued. - assertThatThrownBy(() -> channel.toInputChannel(false)) + assertThatThrownBy(() -> channel.toInputChannel(true)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Received buffer should be empty"); @@ -92,34 +108,11 @@ class RecoveredInputChannelTest { channel.onRecoveredStateBuffer(BufferBuilderTestUtils.buildSomeBuffer()); channel.finishReadRecoveredState(); - assertThatThrownBy(() -> channel.toInputChannel(false)) + assertThatThrownBy(() -> channel.toInputChannel(true)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Received buffer should be empty"); } - @Test - void testToInputChannelPushesQueuedBuffersWhenNeedsRecovery() throws IOException { - // FLINK-38544 transitional: removed when the spilling backend lands (recovered state then - // goes to disk, the queue is always empty at conversion, and toInputChannel(true) asserts - // emptiness instead of pushing). - TestableRecoveredInputChannel channel = buildTestableChannel(true); - - channel.onRecoveredStateBuffer(BufferBuilderTestUtils.buildSomeBuffer(42)); - channel.finishReadRecoveredState(); - - TestInputChannel converted = (TestInputChannel) channel.toInputChannel(true); - - // The queued data buffer is handed over through the push interface and the legacy - // EndOfInputChannelStateEvent is dropped in translation. The EndOfFetchedChannelStateEvent - // sentinel is deliberately NOT appended here: the StreamTask recovery chain appends it via - // finishRecoveredBufferDelivery() on the channel IO executor after partitions have been - // requested (the sentinel must not become consumable before upstream readiness). - assertThat(converted.getRecoveredBuffersSpy()).hasSize(1); - Buffer data = converted.getRecoveredBuffersSpy().pollFirst(); - assertThat(data.isBuffer()).isTrue(); - assertThat(data.getSize()).isEqualTo(42); - } - @Test void testStateConsumedFutureCompletesAfterLegacySentinelIsConsumed() throws IOException { RecoveredInputChannel channel = buildChannel(false); @@ -166,6 +159,84 @@ class RecoveredInputChannelTest { } } + @Test + void testBufferPoolExhaustedBlocksRatherThanHeapAllocate() throws Exception { + int totalSegments = 4; + pool = new NetworkBufferPool(totalSegments, MemoryManager.DEFAULT_PAGE_SIZE); + RecoveredInputChannel channel = buildPooledChannel(pool, totalSegments); + + for (int i = 0; i < totalSegments; i++) { + channel.requestBufferBlocking(false); + } + + CountDownLatch entered = new CountDownLatch(1); + AtomicReference<Buffer> result = new AtomicReference<>(); + Thread blocker = + new Thread( + () -> { + try { + entered.countDown(); + result.set(channel.requestBufferBlocking(false)); + } catch (Exception ignored) { + // Thread will be interrupted at teardown. + } + }, + "blocking-requester"); + blocker.start(); + + assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); + assertThat(result.get()).as("buffer should not have been allocated").isNull(); + + blocker.interrupt(); + blocker.join(5_000); + } + + @Test + void testFilterOnPathTakesSameRouteAsFilterOff() throws Exception { + int exclusivePerChannel = 1; + int totalSegments = 4; + pool = new NetworkBufferPool(totalSegments, MemoryManager.DEFAULT_PAGE_SIZE); + + Buffer filterOnBuf = + buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking(true); + Buffer filterOffBuf = + buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking(false); + + // Both must come from the pool — the BufferManager-owned recycler, not the + // FreeingBufferRecycler the heap-fallback used. + assertThat(filterOnBuf.getMemorySegment()).isNotNull(); + assertThat(filterOffBuf.getMemorySegment()).isNotNull(); + assertThat(filterOnBuf.getRecycler().getClass().getName()) + .doesNotContain("FreeingBufferRecycler"); + assertThat(filterOffBuf.getRecycler().getClass().getName()) + .doesNotContain("FreeingBufferRecycler"); + + filterOnBuf.recycleBuffer(); + filterOffBuf.recycleBuffer(); + } + + private RecoveredInputChannel buildPooledChannel( + NetworkBufferPool segmentProvider, int exclusivePerChannel) { + SingleInputGate inputGate = + new SingleInputGateBuilder().setSegmentProvider(segmentProvider).build(); + return new RecoveredInputChannel( + inputGate, + 0, + new ResultPartitionID(), + new ResultSubpartitionIndexSet(0), + 0, + 0, + new SimpleCounter(), + new SimpleCounter(), + exclusivePerChannel) { + @Override + protected InputChannel toInputChannelInternal(boolean needsRecovery) { + throw new AssertionError("not expected during this test"); + } + }; + } + /** * A RecoveredInputChannel that returns a TestInputChannel when converted, for testing purposes. */ diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java index 7aa0461e0d9..afa6d6fc6d8 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteRecoveredInputChannelTest.java @@ -41,7 +41,7 @@ class RemoteRecoveredInputChannelTest { try { recoveredChannel.finishReadRecoveredState(); - assertThatThrownBy(() -> recoveredChannel.toInputChannel(false)) + assertThatThrownBy(() -> recoveredChannel.toInputChannel(true)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("Received buffer should be empty"); } finally {
