You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2026/08/13 21:15:17 UTC
(flink) 03/04: [FLINK-40080] Switch flag-on recovery to fetch->drain; retire the in-memory backend
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 <19...@gmail.com>
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 {