This is an automated email from the ASF dual-hosted git repository. 1996fanrui pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit ad50a3072837a3fac658f1bab7b021bf62591165 Author: Rui Fan <[email protected]> AuthorDate: Thu Jul 2 00:53:05 2026 +0200 [FLINK-40016][checkpoint] Prevent recovered buffers from being persisted twice in LocalInputChannel --- .../partition/consumer/LocalInputChannel.java | 7 ++-- .../partition/consumer/LocalInputChannelTest.java | 48 ++++++++++++++++++++++ 2 files changed, 52 insertions(+), 3 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 55ff29619c4..f7787c85c6c 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 @@ -332,7 +332,10 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit return Optional.of(getBufferAndAvailability(toBeConsumedBuffers.removeFirst())); } - return Optional.of(getBufferAndAvailability(next)); + BufferAndAvailability bufferAndAvailability = getBufferAndAvailability(next); + channelStatePersister.checkForBarrier(bufferAndAvailability.buffer()); + channelStatePersister.maybePersist(bufferAndAvailability.buffer()); + return Optional.of(bufferAndAvailability); } /** @@ -411,8 +414,6 @@ public class LocalInputChannel extends InputChannel implements BufferAvailabilit numBytesIn.inc(buffer.readableBytes()); numBuffersIn.inc(); - channelStatePersister.checkForBarrier(buffer); - channelStatePersister.maybePersist(buffer); NetworkActionsLogger.traceInput( "LocalInputChannel#getNextBuffer", buffer, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java index 86bda9866d2..44c4b48cb7a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannelTest.java @@ -793,6 +793,54 @@ class LocalInputChannelTest { .containsExactly(10, 20, 30); } + @Test + void testRecoveredBuffersNotPersistedAgainWhenConsumedDuringCheckpoint() throws Exception { + // given: Channel with 2 recovered buffers + SingleInputGate inputGate = new SingleInputGateBuilder().build(); + + ArrayDeque<Buffer> recoveredBuffers = new ArrayDeque<>(); + recoveredBuffers.add(TestBufferFactory.createBuffer(10)); + recoveredBuffers.add(TestBufferFactory.createBuffer(20)); + + RecordingChannelStateWriter stateWriter = new RecordingChannelStateWriter(); + + LocalInputChannel channel = + new LocalInputChannel( + inputGate, + 0, + new ResultPartitionID(), + new ResultSubpartitionIndexSet(0), + new ResultPartitionManager(), + new TaskEventDispatcher(), + 0, + 0, + new SimpleCounter(), + new SimpleCounter(), + stateWriter, + recoveredBuffers); + + inputGate.setInputChannels(channel); + + // when: Checkpoint starts — checkpointStarted spills recovered buffers as inflight data + CheckpointOptions options = + CheckpointOptions.unaligned(CheckpointType.CHECKPOINT, getDefault()); + stateWriter.start(1L, options); + channel.checkpointStarted(new CheckpointBarrier(1L, 0L, options)); + + // then: exactly 2 buffers persisted so far + assertThat(stateWriter.getAddedInput().get(channel.getChannelInfo())).hasSize(2); + + // when: Consume the recovered buffers while checkpoint is still in BARRIER_PENDING state + channel.getNextBuffer(); + channel.getNextBuffer(); + + // then: total persisted count must still be 2 — recovered buffers must not be + // re-persisted via maybePersist() when consumed (that would make it 4 without the fix) + assertThat(stateWriter.getAddedInput().get(channel.getChannelInfo())) + .as("recovered buffers must not be persisted again when consumed") + .hasSize(2); + } + @Test void testPriorityEventConsumedBeforeRecoveredBuffers() throws Exception { RecordingChannelStateWriter stateWriter = new RecordingChannelStateWriter();
