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 c09a21c82bc15ae9b515c49bbffee64f11ea6311 Author: Rui Fan <[email protected]> AuthorDate: Thu May 21 15:26:09 2026 +0200 [FLINK-39520][network] Decouple LocalInputChannel recovery wiring from toBeConsumedBuffers Split the single toBeConsumedBuffers queue into two queues with disjoint responsibilities: - recoveredBuffers (new): holds buffers migrated from RecoveredInputChannel during construction; consumed by getNextRecoveredBuffer() which retains the priority-event interleaving and last-buffer dynamic next-data-type detection introduced by FLINK-39018. - toBeConsumedBuffers (existing): reverted to its pre-FLINK-39018 role of holding FullyFilledBuffer partial-buffer splits only. The recovery-aware early branch in getNextBuffer() and the checkpointStarted inflight scan no longer touch this queue. Restores the checkState(toBeConsumedBuffers.isEmpty()) guard in requestSubpartitions() (removed by cebc174a). hasPendingPriorityEvent, notifyPriorityEvent, and the constructor signature are unchanged. Pure refactor: no public API change, no new tests; verified by the 9 existing LocalInputChannelTest regression cases. --- .../partition/consumer/LocalInputChannel.java | 63 ++++++++++++++-------- 1 file changed, 42 insertions(+), 21 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java index f7787c85c6c..001933a04c8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java @@ -80,9 +80,16 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit private final Deque<BufferAndBacklog> toBeConsumedBuffers = new ArrayDeque<>(); + /** + * Buffers migrated from {@code RecoveredInputChannel}, kept separately from {@link + * #toBeConsumedBuffers} so that recovery semantics (priority event interleaving, checkpoint + * inflight persistence) do not leak into the FullyFilledBuffer split path. + */ + private final Deque<BufferAndBacklog> recoveredBuffers = new ArrayDeque<>(); + /** * Flag indicating whether there is a pending priority event (e.g., checkpoint barrier) in the - * subpartitionView that should be consumed before toBeConsumedBuffers. This is set by {@link + * subpartitionView that should be consumed before recoveredBuffers. This is set by {@link * #notifyPriorityEvent} and checked in {@link #getNextBuffer()}. */ private volatile boolean hasPendingPriorityEvent = false; @@ -131,13 +138,13 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit // buffersInBacklog is set to 0 as these are recovered buffers BufferAndBacklog bufferAndBacklog = new BufferAndBacklog(buffer, 0, nextDataType, seqNum++); - toBeConsumedBuffers.add(bufferAndBacklog); + recoveredBuffers.add(bufferAndBacklog); } checkState( - toBeConsumedBuffers.size() == expectedCount, + recoveredBuffers.size() == expectedCount, "Buffer migration failed: expected %s buffers but got %s", expectedCount, - toBeConsumedBuffers.size()); + recoveredBuffers.size()); } } @@ -146,10 +153,11 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit // ------------------------------------------------------------------------ public void checkpointStarted(CheckpointBarrier barrier) throws CheckpointException { - // Collect inflight buffers from toBeConsumedBuffers to be persisted. - // These are buffers that have not been consumed yet when the checkpoint barrier arrives. + // Collect inflight buffers from recoveredBuffers to be persisted. + // These are recovered buffers that have not been consumed yet when the checkpoint + // barrier arrives. List<Buffer> inflightBuffers = new ArrayList<>(); - for (BufferAndBacklog bufferAndBacklog : toBeConsumedBuffers) { + for (BufferAndBacklog bufferAndBacklog : recoveredBuffers) { if (bufferAndBacklog.buffer().isBuffer()) { inflightBuffers.add(bufferAndBacklog.buffer().retainBuffer()); } @@ -163,6 +171,8 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit @Override protected void requestSubpartitions() throws IOException { + checkState(toBeConsumedBuffers.isEmpty()); + boolean retriggerRequest = false; boolean notifyDataAvailable = false; @@ -272,10 +282,14 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit public Optional<BufferAndAvailability> getNextBuffer() throws IOException { checkError(); - if (!toBeConsumedBuffers.isEmpty()) { + if (!recoveredBuffers.isEmpty()) { return getNextRecoveredBuffer(); } + if (!toBeConsumedBuffers.isEmpty()) { + return Optional.of(getBufferAndAvailability(toBeConsumedBuffers.removeFirst())); + } + ResultSubpartitionView subpartitionView = this.subpartitionView; if (subpartitionView == null) { // There is a possible race condition between writing a EndOfPartitionEvent (1) and @@ -339,12 +353,12 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit } /** - * Consumes the next buffer from toBeConsumedBuffers (recovered buffers), handling pending - * priority events and dynamic availability detection for the last recovered buffer. + * Consumes the next buffer from recoveredBuffers, handling pending priority events and dynamic + * availability detection for the last recovered buffer. */ private Optional<BufferAndAvailability> getNextRecoveredBuffer() throws IOException { // If there is a pending priority event (e.g., unaligned checkpoint barrier), fetch it - // from subpartitionView first, skipping toBeConsumedBuffers. This ensures priority + // from subpartitionView first, skipping recoveredBuffers. This ensures priority // events are processed immediately even when there are pending recovered buffers. if (hasPendingPriorityEvent) { checkState(subpartitionView != null, "No subpartition view available"); @@ -362,10 +376,10 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit if (!expectedNextDataType.hasPriority()) { // Reset hasPendingPriorityEvent to false if no more priority event hasPendingPriorityEvent = false; - if (!toBeConsumedBuffers.isEmpty()) { - // Correct nextDataType: if toBeConsumedBuffers is not empty, the actual next - // element to consume is from toBeConsumedBuffers, not from subpartitionView - expectedNextDataType = toBeConsumedBuffers.peek().buffer().getDataType(); + if (!recoveredBuffers.isEmpty()) { + // Correct nextDataType: if recoveredBuffers is not empty, the actual next + // element to consume is from recoveredBuffers, not from subpartitionView + expectedNextDataType = recoveredBuffers.peek().buffer().getDataType(); } } @@ -378,13 +392,13 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit next.getSequenceNumber()))); } - BufferAndBacklog next = toBeConsumedBuffers.removeFirst(); + BufferAndBacklog next = recoveredBuffers.removeFirst(); // If this is the last recovered buffer and nextDataType is NONE, // dynamically check if subpartitionView has data available. // The last buffer's nextDataType was preset to NONE during construction, // but subpartitionView may already have data available. - if (toBeConsumedBuffers.isEmpty() + if (recoveredBuffers.isEmpty() && next.getNextDataType() == Buffer.DataType.NONE && subpartitionView != null) { ResultSubpartitionView.AvailabilityWithBacklog availability = @@ -510,8 +524,13 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit subpartitionView = null; } - // Release any remaining buffers in toBeConsumedBuffers to avoid memory leak. - // These may be recovered buffers or partial buffers from FullyFilledBuffer. + // Release any remaining buffers in recoveredBuffers (migrated recovered buffers + // not yet consumed) and toBeConsumedBuffers (FullyFilledBuffer partial splits) + // to avoid memory leak. + for (BufferAndBacklog bufferAndBacklog : recoveredBuffers) { + bufferAndBacklog.buffer().recycleBuffer(); + } + recoveredBuffers.clear(); for (BufferAndBacklog bufferAndBacklog : toBeConsumedBuffers) { bufferAndBacklog.buffer().recycleBuffer(); } @@ -532,14 +551,16 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit @Override int getBuffersInUseCount() { ResultSubpartitionView view = this.subpartitionView; - return toBeConsumedBuffers.size() + (view == null ? 0 : view.getNumberOfQueuedBuffers()); + return recoveredBuffers.size() + + toBeConsumedBuffers.size() + + (view == null ? 0 : view.getNumberOfQueuedBuffers()); } @Override public int unsynchronizedGetNumberOfQueuedBuffers() { ResultSubpartitionView view = subpartitionView; - int count = toBeConsumedBuffers.size(); + int count = recoveredBuffers.size() + toBeConsumedBuffers.size(); if (view != null) { count += view.unsynchronizedGetNumberOfQueuedBuffers(); }
