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:14 UTC
(flink) branch master updated (41f2cdb8864 -> d9f5e05ea7a)
This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 41f2cdb8864 [FLINK-40386][checkpoint] Log which channel state recovery path is taken and its progress
new ddeb17e52ce [FLINK-40080][checkpoint] ChannelStateWriter#addInputDataFromSpill: replay spilled segments into the checkpoint
new 46ad84a6133 [FLINK-40080][checkpoint] FetchedChannelStateDrainer: incremental drain with atomic snapshot-and-insert
new af146b61707 [FLINK-40080] Switch flag-on recovery to fetch->drain; retire the in-memory backend
new d9f5e05ea7a [FLINK-40080][network] Remove the unbounded heap fallback from RecoveredInputChannel#requestBufferBlocking
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../channel/ChannelStateCheckpointWriter.java | 47 ++
.../channel/ChannelStateFilteringHandler.java | 225 -------
.../checkpoint/channel/ChannelStateSerializer.java | 14 +
.../channel/ChannelStateWriteRequest.java | 14 +
.../checkpoint/channel/ChannelStateWriter.java | 20 +
.../checkpoint/channel/ChannelStateWriterImpl.java | 7 +
.../channel/FetchedChannelStateDrainer.java | 167 +++++-
.../channel/RecoveredChannelStateHandler.java | 191 +-----
.../channel/SequentialChannelStateReader.java | 2 -
.../channel/SequentialChannelStateReaderImpl.java | 76 +--
.../partition/consumer/RecoveredInputChannel.java | 79 +--
.../runtime/io/checkpointing/ChannelState.java | 18 +-
.../io/checkpointing/InputProcessorUtil.java | 9 +-
.../SingleCheckpointBarrierHandler.java | 9 +-
.../channel/ChannelStateWriterImplTest.java | 119 ++++
.../channel/FetchedChannelStateDrainerTest.java | 663 +++++++++++++++++++++
.../InputChannelRecoveredStateHandlerTest.java | 365 +++++++++---
.../checkpoint/channel/MockChannelStateWriter.java | 10 +
.../consumer/LocalRecoveredInputChannelTest.java | 2 +-
.../consumer/RecoveredInputChannelTest.java | 119 +++-
.../consumer/RemoteRecoveredInputChannelTest.java | 2 +-
.../runtime/io/checkpointing/ChannelStateTest.java | 124 +++-
.../checkpointing/TestBarrierHandlerFactory.java | 1 +
23 files changed, 1604 insertions(+), 679 deletions(-)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainerTest.java
(flink) 02/04: [FLINK-40080][checkpoint] FetchedChannelStateDrainer: incremental drain with atomic snapshot-and-insert
Posted by fa...@apache.org.
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 46ad84a6133b0b34fbc9cc622d2124de007286d5
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 02:15:24 2026 +0200
[FLINK-40080][checkpoint] FetchedChannelStateDrainer: incremental drain with atomic snapshot-and-insert
Implements RecoveryCheckpointTrigger + Closeable; the constructor takes
a FetchedChannelState and the task's List<RecoverableInputChannel> and
derives the InputChannelInfo map internally.
drain() walks the root reader front to back; per segment the recovery
buffer request (requestRecoveryBufferBlocking()) and the disk read
happen outside the lock, while delivery (onRecoveredStateBuffer) and
commit()/offset advance happen inside it; at the end
finishRecoveredBufferDelivery() is called per channel.
snapshotAndInsertBarriers(cpId) is atomic under the drainer lock:
snapshot the committed position (FetchedChannelStateSnapshot,
ref-counted acquire) and insert a RecoveryCheckpointBarrier into every
in-recovery channel; returns the snapshot's reader (caller-owned).
Nothing constructs a drainer in production yet and the handler factory
is unswitched, so runtime behavior is unchanged.
---
.../channel/FetchedChannelStateDrainer.java | 167 +++++-
.../channel/SequentialChannelStateReader.java | 2 -
.../channel/SequentialChannelStateReaderImpl.java | 76 +--
.../channel/FetchedChannelStateDrainerTest.java | 663 +++++++++++++++++++++
4 files changed, 834 insertions(+), 74 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java
index 23f33bca40e..0fe18769c62 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainer.java
@@ -18,71 +18,182 @@
package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.consumer.RecoverableInputChannel;
import java.io.Closeable;
import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
/**
- * Drains a {@link FetchedChannelState} into the physical recovered channels and inserts {@link
- * RecoveryCheckpointBarrier}s when a checkpoint fires during recovery.
+ * Drains a {@link FetchedChannelState} into recovered-buffer queues and snapshots remaining
+ * segments when a checkpoint fires during recovery.
*
- * <p>FLINK-38544 transitional in-memory implementation: the in-memory recovery backend already
- * pushed every recovered buffer into the physical channels' own queues at conversion time (via
- * {@code requestPartitions(true)}), so draining is just appending the end-of-recovered-state
- * sentinel to each channel, and there is never an undrained residue to snapshot — inserting the
- * barrier into the in-recovery channels is enough. The disk-based drainer of the spilling backend
- * replaces this, reading segments off spill files and returning a reader over the undrained slice.
+ * <p>The drainer lock pairs channel delivery with reader-cursor advancement and also protects
+ * snapshot creation plus barrier insertion. Disk reads and buffer allocation stay outside that
+ * lock.
*/
@Internal
public final class FetchedChannelStateDrainer implements RecoveryCheckpointTrigger, Closeable {
+ private final FetchedChannelStateReader rootReader;
+
+ private final ResolvedChannels channels;
+
+ private final Object lock = new Object();
private final FetchedChannelState channelState;
- private final List<RecoverableInputChannel> channels;
+ /**
+ * Set under {@link #lock} once {@link #drain()} has consumed every segment. After that the
+ * {@link #rootReader} is closed by {@link #close()}, so a later {@link
+ * #snapshotAndInsertBarriers} must not derive from it; it returns an empty reader instead.
+ * Guarded by the lock so the check is atomic with barrier insertion.
+ */
+ private boolean drainFinished;
public FetchedChannelStateDrainer(
FetchedChannelState channelState, List<RecoverableInputChannel> channels) {
- this.channelState = checkNotNull(channelState);
- this.channels = checkNotNull(channels);
+ this.channelState = channelState;
+ this.rootReader = checkNotNull(channelState).reader();
+ this.channels = new ResolvedChannels(channels);
+ }
+
+ private static final class ResolvedChannels {
+ final List<RecoverableInputChannel> allChannels;
+ final Map<InputChannelInfo, RecoverableInputChannel> channelByInfo;
+
+ ResolvedChannels(List<RecoverableInputChannel> all) {
+ this.allChannels = all;
+ Map<InputChannelInfo, RecoverableInputChannel> byInfo = new HashMap<>();
+ for (RecoverableInputChannel ch : all) {
+ byInfo.put(ch.getChannelInfo(), ch);
+ }
+ this.channelByInfo = byInfo;
+ }
}
/**
- * Appends the end-of-recovered-state sentinel to every converted channel. Each channel first
- * waits for its upstream to be ready, so this must run on the channelIOExecutor rather than
- * block the mailbox thread. Only once the sentinel is in place can the consume path flip the
- * channel out of recovery, which guarantees live data is never polled before the upstream
- * connection exists.
+ * Drains all segments from the spill file into the corresponding recovery buffer queues. Each
+ * segment is split into chunks of at most {@code memorySegmentSize} bytes; a full chunk is
+ * delivered under the drainer lock paired with a segment commit. After all segments are
+ * drained, every channel's {@link RecoverableInputChannel#finishRecoveredBufferDelivery()} is
+ * called.
+ *
+ * <p>Disk reads and buffer allocations happen outside the lock; only the "deliver + commit"
+ * pair is locked to guarantee atomicity with snapshot.
*/
public void drain() throws IOException, InterruptedException {
channelState.release();
- for (RecoverableInputChannel channel : channels) {
- channel.finishRecoveredBufferDelivery();
+ Optional<SpillSegment> next;
+ while ((next = rootReader.advanceAndGetNextSegment()).isPresent()) {
+ SpillSegment seg = next.get();
+ RecoverableInputChannel ch = channels.channelByInfo.get(seg.channelInfo());
+ if (ch == null) {
+ throw new IllegalStateException(
+ "Drain: no physical channel found for " + seg.channelInfo());
+ }
+ drainSegment(seg, ch);
+ }
+
+ // Mark drain done before rootReader is closed, so a concurrent snapshot returns empty
+ // rather than deriving from the soon-to-be-closed rootReader. Under the lock to stay atomic
+ // with snapshotAndInsertBarriers' check.
+ synchronized (lock) {
+ drainFinished = true;
+ }
+ for (RecoverableInputChannel ch : channels.allChannels) {
+ ch.finishRecoveredBufferDelivery();
}
}
/**
- * Inserts a {@link RecoveryCheckpointBarrier} into every channel that is still in recovery, so
- * that {@code checkpointStarted}'s in-recovery branch can persist exactly the pre-barrier
- * recovered data. There is no snapshot side for the in-memory backend: everything a checkpoint
- * must persist is already inside the channels' queues, so the snapshot is inherently empty.
+ * Drains one segment into the given channel, delivering a buffer under the lock once it is full
+ * or the segment is exhausted.
+ *
+ * <p>{@link RecoverableInputChannel#onRecoveredStateBuffer} takes ownership even when it
+ * throws, so the reference is dropped before the call; whatever is still held is recycled on
+ * error.
+ */
+ private void drainSegment(SpillSegment seg, RecoverableInputChannel ch)
+ throws IOException, InterruptedException {
+ InputStream in = seg.bodyStream();
+ int remaining = seg.length();
+ Buffer buf = null;
+ try {
+ while (remaining > 0) {
+ if (buf == null) {
+ buf = ch.requestRecoveryBufferBlocking();
+ }
+ remaining -=
+ fill(buf, in, Math.min(buf.getMaxCapacity() - buf.getSize(), remaining));
+ if (buf.getSize() == buf.getMaxCapacity() || remaining == 0) {
+ Buffer delivered = buf;
+ buf = null;
+ synchronized (lock) {
+ ch.onRecoveredStateBuffer(delivered);
+ seg.commit();
+ }
+ }
+ }
+ } catch (Throwable t) {
+ if (buf != null) {
+ buf.recycleBuffer();
+ }
+ throw t;
+ }
+ }
+
+ /**
+ * Writes up to {@code toRead} bytes from {@code in} into {@code buf} and returns how many were
+ * written. Does not close or recycle {@code buf}; ownership stays with the caller.
+ */
+ private static int fill(Buffer buf, InputStream in, int toRead) throws IOException {
+ checkArgument(toRead > 0);
+ // Do not use try-with-resources: ChannelStateByteBuffer.close() recycles the buffer,
+ // but the buffer is still owned by the caller here.
+ ChannelStateByteBuffer view = ChannelStateByteBuffer.wrap(buf);
+ return view.writeBytes(in, toRead);
+ }
+
+ /**
+ * Atomically snapshots the undrained portion of the spill and inserts {@link
+ * RecoveryCheckpointBarrier}s into all in-recovery channels. Returns an independent reader over
+ * the remaining segments for replay into the checkpoint stream; the caller owns and must close
+ * it.
+ *
+ * <p>If the drain has already finished, the root reader is closed and there is nothing left to
+ * snapshot; an empty reader is returned so the caller's normal flow handles it uniformly.
*/
@Override
public FetchedChannelStateSnapshot snapshotAndInsertBarriers(long checkpointId)
throws IOException {
- for (RecoverableInputChannel channel : channels) {
- channel.insertRecoveryCheckpointBarrierIfInRecovery(checkpointId);
+
+ // Barrier insertion and snapshot must occur within the same critical section so that the
+ // snapshot's committed position reflects exactly the drain position at the moment barriers
+ // were inserted, with no window for the drain thread to advance between.
+ synchronized (lock) {
+ for (RecoverableInputChannel ch : channels.allChannels) {
+ ch.insertRecoveryCheckpointBarrierIfInRecovery(checkpointId);
+ }
+ if (drainFinished) {
+ // Drain consumed everything and rootReader is (being) closed; nothing left to
+ // snapshot. Return an empty snapshot so the caller's normal flow handles it.
+ return FetchedChannelState.emptySnapshot();
+ }
+ return rootReader.snapshot();
}
- // No snapshot side for the in-memory backend: everything a checkpoint must persist is
- // already inside the channels' queues, so the returned snapshot is empty.
- return FetchedChannelState.emptySnapshot();
}
@Override
public void close() throws IOException {
- channelState.close();
+ rootReader.close();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java
index f6b8decf09c..88296c517b0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReader.java
@@ -34,8 +34,6 @@ public interface SequentialChannelStateReader extends AutoCloseable {
*
* @param inputGates The input gates to recover state for.
* @param filterContext The filter context containing input configs and rescaling info.
- * @return the fetched recovered channel state if there is any state to recover, otherwise
- * {@link Optional#empty()}.
*/
Optional<FetchedChannelState> readInputData(
InputGate[] inputGates, RecordFilterContext filterContext)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
index 784aa7c1c0d..9284c226a59 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java
@@ -33,7 +33,6 @@ import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext;
import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -72,45 +71,38 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR
? ChannelStateFilteringHandler.createFromContext(filterContext, inputGates)
: null;
- try (ChannelStateFilteringHandler ignored = filteringHandler;
- AbstractInputChannelRecoveredStateHandler stateHandler =
- AbstractInputChannelRecoveredStateHandler.create(
- inputGates,
- taskStateSnapshot.getInputRescalingDescriptor(),
- filterContext.isCheckpointingDuringRecoveryEnabled(),
- filteringHandler,
- filterContext.getMemorySegmentSize(),
- filterContext.getTmpDirectories())) {
- boolean readAny =
- read(
- stateHandler,
- groupByDelegate(
- streamSubtaskStates(),
- ChannelStateHelper::extractUnmergedInputHandles));
- readAny |=
- read(
- stateHandler,
- groupByDelegate(
- streamSubtaskStates(),
- OperatorSubtaskState::getUpstreamOutputBufferState));
-
- if (filteringHandler != null) {
- checkState(
- !filteringHandler.hasPartialData(),
- "Not all data has been fully consumed during filtering");
+ // Manual close ordering so the produced spill file can be published after
+ // stateHandler.close() flushes the filter writer.
+ AbstractInputChannelRecoveredStateHandler stateHandler =
+ AbstractInputChannelRecoveredStateHandler.create(
+ inputGates,
+ taskStateSnapshot.getInputRescalingDescriptor(),
+ filterContext.isCheckpointingDuringRecoveryEnabled(),
+ filteringHandler,
+ filterContext.getMemorySegmentSize(),
+ filterContext.getTmpDirectories());
+ try (ChannelStateFilteringHandler ignored = filteringHandler) {
+ try (stateHandler) {
+ read(
+ stateHandler,
+ groupByDelegate(
+ streamSubtaskStates(),
+ ChannelStateHelper::extractUnmergedInputHandles));
+ read(
+ stateHandler,
+ groupByDelegate(
+ streamSubtaskStates(),
+ OperatorSubtaskState::getUpstreamOutputBufferState));
+
+ if (filteringHandler != null) {
+ checkState(
+ !filteringHandler.hasPartialData(),
+ "Not all data has been fully consumed during filtering");
+ }
}
- // The container signals "any recovered data was pushed", not "a filter ran": the
- // no-checkpointing path pushes recovered buffers directly and must produce nothing
- // here, matching the caller's checkState(readInputData(...).isEmpty()).
- //
- // FLINK-38544 transitional in-memory backend: recovered buffers already live in the
- // physical channels' queues, so the returned container is an empty placeholder.
- // TODO: FLINK-38544 — return stateHandler.getProducedChannelState() once the handler
- // factory selects the Spilling* handlers; as-is their file-backed state would be
- // silently dropped here.
- return filterContext.isCheckpointingDuringRecoveryEnabled() && readAny
- ? Optional.of(new FetchedChannelState(Collections.emptyList()))
- : Optional.empty();
+ // stateHandler.close() (above) has flushed the filter writer and published the
+ // produced spill file, so read getProducedChannelState() after the close completes.
+ return Optional.ofNullable(stateHandler.getProducedChannelState());
}
}
@@ -130,19 +122,15 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR
}
}
- /** Returns {@code true} if any channel state handle was read. */
- private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> boolean read(
+ private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> void read(
RecoveredChannelStateHandler<Info, Context> stateHandler,
Map<StreamStateHandle, List<Handle>> streamStateHandleListMap)
throws IOException, InterruptedException {
- boolean readAny = false;
for (Map.Entry<StreamStateHandle, List<Handle>> delegateAndHandles :
streamStateHandleListMap.entrySet()) {
readSequentially(
delegateAndHandles.getKey(), delegateAndHandles.getValue(), stateHandler);
- readAny = true;
}
- return readAny;
}
private <Info, Context, Handle extends AbstractChannelStateHandle<Info>> void readSequentially(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainerTest.java
new file mode 100644
index 00000000000..d729bef9e98
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateDrainerTest.java
@@ -0,0 +1,663 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.checkpoint.channel;
+
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment;
+import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
+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.partition.consumer.RecoverableInputChannel;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for {@link FetchedChannelStateDrainer}: drain demux, finish ordering, snapshot start
+ * position, barrier insertion, and edge cases (drain-finished, channel-not-in-recovery).
+ */
+class FetchedChannelStateDrainerTest {
+
+ @TempDir Path tempDir;
+
+ @Test
+ void testDrainEndToEnd() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1), payload(2), payload(3));
+
+ RecordingChannel rec = new RecordingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, rec);
+
+ drainer.drain();
+ drainer.close();
+
+ // All segment bodies must be delivered as buffer(s); at least 3 non-empty deliveries
+ // because the segment body contains 3 records but they may be batched into fewer buffers.
+ assertThat(rec.recovered).isNotEmpty();
+ assertThat(rec.finishCalls).isEqualTo(1);
+ }
+
+ @Test
+ void testDrainSegmentLargerThanBufferSplitsIntoFullChunksThenPartialTail() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+
+ // Buffer capacity deliberately smaller than the segment body so the drainer must fill
+ // multiple buffers and a final partial tail. 50 bytes over a 16-byte buffer => 16+16+16+2.
+ int bufferCapacity = 16;
+ byte[] body = sequentialBytes(50);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ // Pass-through so the segment body equals the verbatim bytes (no length framing).
+ writer.writePassThrough(cInfo, body, 0, body.length);
+ state = writer.getChannelState();
+ }
+
+ RecordingChannel rec = new RecordingChannel(cInfo, bufferCapacity);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, rec);
+
+ drainer.drain();
+ drainer.close();
+
+ // ceil(50 / 16) = 4 buffers delivered.
+ assertThat(rec.recovered).hasSize(4);
+ // Every buffer except the last is filled to capacity; the last carries the remainder.
+ for (int i = 0; i < rec.recovered.size() - 1; i++) {
+ assertThat(rec.recovered.get(i).getSize()).isEqualTo(bufferCapacity);
+ }
+ assertThat(rec.recovered.get(rec.recovered.size() - 1).getSize())
+ .isEqualTo(body.length % bufferCapacity);
+
+ // Buffers concatenated in delivery order must reproduce the segment body byte-for-byte.
+ assertThat(concat(rec.recovered)).isEqualTo(body);
+ assertThat(rec.finishCalls).isEqualTo(1);
+ }
+
+ @Test
+ void testDrainSegmentExactMultipleOfBufferHasNoPartialTail() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+
+ // Body length is an exact multiple of the buffer capacity: the segment ends on a buffer
+ // boundary, so no extra buffer is requested and no empty buffer is delivered.
+ int bufferCapacity = 16;
+ byte[] body = sequentialBytes(bufferCapacity * 3);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writePassThrough(cInfo, body, 0, body.length);
+ state = writer.getChannelState();
+ }
+
+ RecordingChannel rec = new RecordingChannel(cInfo, bufferCapacity);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, rec);
+
+ drainer.drain();
+ drainer.close();
+
+ // Exactly 3 full buffers, no trailing empty buffer.
+ assertThat(rec.recovered).hasSize(3);
+ for (Buffer b : rec.recovered) {
+ assertThat(b.getSize()).isEqualTo(bufferCapacity);
+ }
+ assertThat(concat(rec.recovered)).isEqualTo(body);
+ assertThat(rec.finishCalls).isEqualTo(1);
+ }
+
+ @Test
+ void testDrainRecyclesInFlightBufferWhenBodyReadFails() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+
+ int bufferCapacity = 16;
+ byte[] body = sequentialBytes(bufferCapacity * 4);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writePassThrough(cInfo, body, 0, body.length);
+ state = writer.getChannelState();
+ }
+
+ // Cut the file short so the body read fails while the drainer still owns a partially
+ // filled buffer: 40 of the 64 body bytes survive => 2 buffers delivered, the third one
+ // holds 8 bytes when the EOF hits and must be recycled rather than leaked.
+ try (FileChannel file = FileChannel.open(state.files().get(0), StandardOpenOption.WRITE)) {
+ file.truncate(file.size() - 24);
+ }
+
+ RecordingChannel rec = new RecordingChannel(cInfo, bufferCapacity);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, rec);
+
+ assertThatThrownBy(drainer::drain).isInstanceOf(EOFException.class);
+ drainer.close();
+
+ assertThat(rec.recovered).hasSize(2);
+ assertThat(rec.requested).isEqualTo(3);
+ assertThat(rec.recycled).isEqualTo(1);
+ }
+
+ @Test
+ void testDrainDemuxByChannelInfo() throws Exception {
+ InputChannelInfo c0 = new InputChannelInfo(0, 0);
+ InputChannelInfo c1 = new InputChannelInfo(0, 1);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writeRecord(c0, payload(11), payload(11).length);
+ writer.writeRecord(c1, payload(22), payload(22).length);
+ writer.writeRecord(c0, payload(33), payload(33).length);
+ writer.writeRecord(c1, payload(44), payload(44).length);
+ state = writer.getChannelState();
+ }
+
+ RecordingChannel chan0 = new RecordingChannel(c0);
+ RecordingChannel chan1 = new RecordingChannel(c1);
+ FetchedChannelStateDrainer drainer = newDrainer(state, c0, chan0, c1, chan1);
+
+ drainer.drain();
+ drainer.close();
+
+ // Each channel must receive some data buffers
+ assertThat(chan0.recovered).isNotEmpty();
+ assertThat(chan1.recovered).isNotEmpty();
+ // Both channels must have finish called
+ assertThat(chan0.finishCalls).isEqualTo(1);
+ assertThat(chan1.finishCalls).isEqualTo(1);
+ }
+
+ @Test
+ void testDrainCallsFinishAfterAllBufferDeliveries() throws Exception {
+ InputChannelInfo c0 = new InputChannelInfo(0, 0);
+ InputChannelInfo c1 = new InputChannelInfo(0, 1);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writeRecord(c0, payload(1), payload(1).length);
+ writer.writeRecord(c1, payload(2), payload(2).length);
+ state = writer.getChannelState();
+ }
+
+ int[] seq = {0};
+ RecordingChannel chan0 = new RecordingChannel(c0, seq);
+ RecordingChannel chan1 = new RecordingChannel(c1, seq);
+ FetchedChannelStateDrainer drainer = newDrainer(state, c0, chan0, c1, chan1);
+
+ drainer.drain();
+ drainer.close();
+
+ int maxDataSeq = Math.max(chan0.maxDataSeq, chan1.maxDataSeq);
+ int minFinishSeq = Math.min(chan0.finishSeq, chan1.finishSeq);
+ assertThat(maxDataSeq).isLessThan(minFinishSeq);
+ }
+
+ @Test
+ void testSnapshotCoversAllSegmentsBeforeDrain() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(5), payload(6));
+
+ RecordingChannel chan = new RecordingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ long cpId = 42L;
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(cpId).reader();
+
+ // Snapshot must cover all segments (at least 1 segment for cInfo).
+ // The sequential reader requires each segment body to be fully consumed before advancing,
+ // mirroring the real consumer (ChannelStateCheckpointWriter#writeInputFromSpill).
+ int count = 0;
+ Optional<SpillSegment> next;
+ while ((next = snap.advanceAndGetNextSegment()).isPresent()) {
+ drainBody(next.get().bodyStream());
+ count++;
+ }
+ snap.close();
+ assertThat(count).isGreaterThan(0);
+ drainer.close();
+ }
+
+ @Test
+ void testSnapshotInsertsBarrierPerChannel() throws Exception {
+ InputChannelInfo c0 = new InputChannelInfo(0, 0);
+ InputChannelInfo c1 = new InputChannelInfo(0, 1);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writeRecord(c0, payload(1), payload(1).length);
+ writer.writeRecord(c1, payload(2), payload(2).length);
+ state = writer.getChannelState();
+ }
+
+ RecordingChannel chan0 = new RecordingChannel(c0);
+ RecordingChannel chan1 = new RecordingChannel(c1);
+ FetchedChannelStateDrainer drainer = newDrainer(state, c0, chan0, c1, chan1);
+
+ long cpId = 7L;
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(cpId).reader();
+ snap.close();
+
+ assertThat(chan0.recovered).hasSize(1);
+ assertThat(chan1.recovered).hasSize(1);
+ assertThat(extractRecoveryBarrierCheckpointId(chan0.recovered.get(0))).isEqualTo(cpId);
+ assertThat(extractRecoveryBarrierCheckpointId(chan1.recovered.get(0))).isEqualTo(cpId);
+ drainer.close();
+ }
+
+ @Test
+ void testSnapshotInsertsBarrierWhenChannelInRecoveryEvenIfDiskSliceEmpty() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1));
+
+ RecordingChannel chan = new RecordingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ drainer.drain();
+ // Drain finished; simulate the channel still in recovery
+ chan.inRecovery = true;
+ int recoveredBefore = chan.recovered.size();
+
+ long cpId = 6L;
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(cpId).reader();
+ assertThat(snap.advanceAndGetNextSegment()).isEmpty();
+ snap.close();
+
+ // Barrier must be inserted even though disk slice is empty
+ assertThat(chan.recovered).hasSize(recoveredBefore + 1);
+ assertThat(extractRecoveryBarrierCheckpointId(chan.recovered.get(recoveredBefore)))
+ .isEqualTo(cpId);
+ drainer.close();
+ }
+
+ @Test
+ void testSnapshotInsertsBarrierOnlyForChannelsStillInRecovery() throws Exception {
+ InputChannelInfo c0 = new InputChannelInfo(0, 0);
+ InputChannelInfo c1 = new InputChannelInfo(0, 1);
+
+ FetchedChannelState state;
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ writer.writeRecord(c0, payload(1), payload(1).length);
+ state = writer.getChannelState();
+ }
+
+ RecordingChannel chan0 = new RecordingChannel(c0);
+ RecordingChannel chan1 = new RecordingChannel(c1);
+ chan1.inRecovery = false;
+
+ FetchedChannelStateDrainer drainer = newDrainer(state, c0, chan0, c1, chan1);
+
+ long cpId = 11L;
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(cpId).reader();
+ snap.close();
+
+ assertThat(chan0.recovered).hasSize(1);
+ assertThat(extractRecoveryBarrierCheckpointId(chan0.recovered.get(0))).isEqualTo(cpId);
+ assertThat(chan1.recovered).isEmpty();
+ drainer.close();
+ }
+
+ @Test
+ void testSnapshotReturnsEmptyWhenDrainFinishedAndNotInRecovery() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1), payload(2));
+
+ RecordingChannel chan = new RecordingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ drainer.drain();
+ chan.inRecovery = false;
+ int recoveredBefore = chan.recovered.size();
+
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(99L).reader();
+ assertThat(snap.advanceAndGetNextSegment()).isEmpty();
+ snap.close();
+
+ // No barrier added since channel left recovery
+ assertThat(chan.recovered).hasSize(recoveredBefore);
+ drainer.close();
+ }
+
+ @Test
+ void testSnapshotAfterDrainerClosedReturnsEmptyWithoutTouchingClosedRootReader()
+ throws Exception {
+ // Mirrors production order: drain() then close() (which closes the root reader) run before
+ // a
+ // late checkpoint fires snapshotAndInsertBarriers. The drain-finished flag must
+ // short-circuit
+ // so the closed root reader is never snapshotted.
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1), payload(2));
+
+ RecordingChannel chan = new RecordingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ drainer.drain();
+ drainer.close();
+
+ FetchedChannelStateReader snap = drainer.snapshotAndInsertBarriers(99L).reader();
+ assertThat(snap.advanceAndGetNextSegment()).isEmpty();
+ snap.close();
+ }
+
+ @Test
+ void testDrainOnExecutorThreadDeliversAndFinishes() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1));
+
+ CapturingChannel chan = new CapturingChannel(cInfo);
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ ExecutorService channelIOExecutor = Executors.newSingleThreadExecutor();
+ try {
+ CompletableFuture<Void> done = new CompletableFuture<>();
+ channelIOExecutor.execute(
+ () -> {
+ try {
+ drainer.drain();
+ done.complete(null);
+ } catch (Throwable t) {
+ done.completeExceptionally(t);
+ } finally {
+ try {
+ drainer.close();
+ } catch (IOException ignore) {
+ }
+ }
+ });
+
+ done.get(5, TimeUnit.SECONDS);
+ assertThat(chan.dataDeliveries).isGreaterThan(0);
+ assertThat(chan.finishCalled).isTrue();
+ } finally {
+ channelIOExecutor.shutdownNow();
+ assertThat(channelIOExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ void testDrainOnExecutorThreadBubblesDeliveryException() throws Exception {
+ InputChannelInfo cInfo = new InputChannelInfo(0, 0);
+ FetchedChannelState state = writeRecords(cInfo, payload(1));
+
+ RecoverableInputChannel chan =
+ new RecoverableInputChannel() {
+ @Override
+ public InputChannelInfo getChannelInfo() {
+ return cInfo;
+ }
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ throw new RuntimeException("boom");
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() {}
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId) {
+ throw new RuntimeException("boom");
+ }
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() {
+ MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(64);
+ return new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE);
+ }
+
+ @Override
+ public void onRecoveredStateConsumed() {}
+ };
+
+ FetchedChannelStateDrainer drainer = newDrainer(state, cInfo, chan);
+
+ CountDownLatch handlerCalled = new CountDownLatch(1);
+ AtomicReference<Throwable> captured = new AtomicReference<>();
+ ExecutorService channelIOExecutor = Executors.newSingleThreadExecutor();
+ try {
+ channelIOExecutor.execute(
+ () -> {
+ try {
+ drainer.drain();
+ } catch (Throwable t) {
+ captured.set(t);
+ handlerCalled.countDown();
+ } finally {
+ try {
+ drainer.close();
+ } catch (IOException ignore) {
+ }
+ }
+ });
+
+ assertThat(handlerCalled.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(captured.get()).isInstanceOf(RuntimeException.class);
+ assertThat(captured.get().getMessage()).isEqualTo("boom");
+ } finally {
+ channelIOExecutor.shutdownNow();
+ assertThat(channelIOExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ // -------------------------------------------------------------------------------------------
+ // Helpers
+ // -------------------------------------------------------------------------------------------
+
+ private FetchedChannelState writeRecords(InputChannelInfo ch, byte[]... payloads)
+ throws IOException {
+ try (TestSpillWriter writer = new TestSpillWriter(tempDir)) {
+ for (byte[] p : payloads) {
+ writer.writeRecord(ch, p, p.length);
+ }
+ return writer.getChannelState();
+ }
+ }
+
+ private FetchedChannelStateDrainer newDrainer(
+ FetchedChannelState state, Object... infoChannelPairs) {
+ List<RecoverableInputChannel> all = new ArrayList<>();
+ for (int i = 0; i < infoChannelPairs.length; i += 2) {
+ all.add((RecoverableInputChannel) infoChannelPairs[i + 1]);
+ }
+ return new FetchedChannelStateDrainer(state, all);
+ }
+
+ private static long extractRecoveryBarrierCheckpointId(Buffer buffer) throws IOException {
+ AbstractEvent event =
+ EventSerializer.fromBuffer(
+ buffer, RecoveryCheckpointBarrier.class.getClassLoader());
+ buffer.setReaderIndex(0);
+ assertThat(event).isInstanceOf(RecoveryCheckpointBarrier.class);
+ return ((RecoveryCheckpointBarrier) event).getCheckpointId();
+ }
+
+ private static byte[] payload(int id) {
+ return new byte[] {(byte) (id & 0xff), (byte) ((id >> 8) & 0xff), (byte) 0xAB, (byte) 0xCD};
+ }
+
+ /** Builds {@code n} bytes whose values count up modulo 256, so order mismatches are visible. */
+ private static byte[] sequentialBytes(int n) {
+ byte[] out = new byte[n];
+ for (int i = 0; i < n; i++) {
+ out[i] = (byte) i;
+ }
+ return out;
+ }
+
+ /** Concatenates the readable bytes of the given buffers in order. */
+ private static byte[] concat(List<Buffer> buffers) {
+ java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+ for (Buffer b : buffers) {
+ java.nio.ByteBuffer nio = b.getNioBufferReadable();
+ byte[] chunk = new byte[nio.remaining()];
+ nio.get(chunk);
+ out.write(chunk, 0, chunk.length);
+ }
+ return out.toByteArray();
+ }
+
+ /** Fully consumes a segment body so the sequential reader may advance to the next segment. */
+ private static void drainBody(InputStream body) throws IOException {
+ byte[] buf = new byte[256];
+ while (body.read(buf) != -1) {
+ // discard
+ }
+ }
+
+ // -------------------------------------------------------------------------------------------
+ // RecordingChannel stub
+ // -------------------------------------------------------------------------------------------
+
+ private static final int DEFAULT_RECOVERY_BUFFER_CAPACITY = 4096;
+
+ private static final class RecordingChannel implements RecoverableInputChannel {
+ private final InputChannelInfo channelInfo;
+ final List<Buffer> recovered = new ArrayList<>();
+ int finishCalls = 0;
+ private final int[] sequence;
+ private final int bufferCapacity;
+ int requested = 0;
+ int recycled = 0;
+ int maxDataSeq = Integer.MIN_VALUE;
+ int finishSeq = -1;
+ boolean inRecovery = true;
+
+ RecordingChannel(InputChannelInfo channelInfo) {
+ this(channelInfo, null, DEFAULT_RECOVERY_BUFFER_CAPACITY);
+ }
+
+ RecordingChannel(InputChannelInfo channelInfo, int[] sharedSequence) {
+ this(channelInfo, sharedSequence, DEFAULT_RECOVERY_BUFFER_CAPACITY);
+ }
+
+ RecordingChannel(InputChannelInfo channelInfo, int bufferCapacity) {
+ this(channelInfo, null, bufferCapacity);
+ }
+
+ RecordingChannel(InputChannelInfo channelInfo, int[] sharedSequence, int bufferCapacity) {
+ this.channelInfo = channelInfo;
+ this.sequence = sharedSequence;
+ this.bufferCapacity = bufferCapacity;
+ }
+
+ @Override
+ public InputChannelInfo getChannelInfo() {
+ return channelInfo;
+ }
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ recovered.add(buffer);
+ if (sequence != null) {
+ maxDataSeq = Math.max(maxDataSeq, ++sequence[0]);
+ }
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() {
+ finishCalls++;
+ if (sequence != null) {
+ finishSeq = ++sequence[0];
+ }
+ }
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId)
+ throws IOException {
+ if (inRecovery) {
+ recovered.add(
+ EventSerializer.toBuffer(
+ new RecoveryCheckpointBarrier(checkpointId), false));
+ }
+ }
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() {
+ requested++;
+ MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(bufferCapacity);
+ return new NetworkBuffer(
+ seg,
+ memorySegment -> {
+ recycled++;
+ FreeingBufferRecycler.INSTANCE.recycle(memorySegment);
+ });
+ }
+
+ @Override
+ public void onRecoveredStateConsumed() {}
+ }
+
+ /** Counts data deliveries and finish for the executor-thread drain tests. */
+ private static final class CapturingChannel implements RecoverableInputChannel {
+ private final InputChannelInfo channelInfo;
+ int dataDeliveries = 0;
+ boolean finishCalled = false;
+
+ CapturingChannel(InputChannelInfo channelInfo) {
+ this.channelInfo = channelInfo;
+ }
+
+ @Override
+ public InputChannelInfo getChannelInfo() {
+ return channelInfo;
+ }
+
+ @Override
+ public void onRecoveredStateBuffer(Buffer buffer) {
+ if (buffer.isBuffer()) {
+ dataDeliveries++;
+ }
+ }
+
+ @Override
+ public void finishRecoveredBufferDelivery() {
+ finishCalled = true;
+ }
+
+ @Override
+ public void insertRecoveryCheckpointBarrierIfInRecovery(long checkpointId) {}
+
+ @Override
+ public Buffer requestRecoveryBufferBlocking() {
+ MemorySegment seg = MemorySegmentFactory.allocateUnpooledSegment(64);
+ return new NetworkBuffer(seg, FreeingBufferRecycler.INSTANCE);
+ }
+
+ @Override
+ public void onRecoveredStateConsumed() {}
+ }
+}
(flink) 04/04: [FLINK-40080][network] Remove the unbounded heap fallback from RecoveredInputChannel#requestBufferBlocking
Posted by fa...@apache.org.
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 d9f5e05ea7ad12c029e77610156cdae6b5fda947
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 02:30:01 2026 +0200
[FLINK-40080][network] Remove the unbounded heap fallback from RecoveredInputChannel#requestBufferBlocking
Back to the plain bufferManager.requestBufferBlocking() form; the OOM
path (and its FLINK-38544 TODO) is gone -- disk spilling supersedes it.
The filtering path no longer competes with itself for network buffers:
the filter output is written to spill files instead of being
re-serialized into freshly requested buffers, so the deadlock the heap
fallback worked around can no longer occur. The handler-side call site
drops the transitional gating parameter accordingly.
---
.../channel/RecoveredChannelStateHandler.java | 4 +--
.../partition/consumer/RecoveredInputChannel.java | 37 ++--------------------
.../consumer/RecoveredInputChannelTest.java | 10 +++---
3 files changed, 7 insertions(+), 44 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java
index 23960227843..03ac7f8309c 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
@@ -143,9 +143,7 @@ abstract class AbstractInputChannelRecoveredStateHandler
public BufferWithContext<Buffer> getBuffer(InputChannelInfo channelInfo)
throws IOException, InterruptedException {
RecoveredInputChannel channel = getMappedChannels(channelInfo);
- // 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);
+ Buffer buffer = channel.requestBufferBlocking();
return new BufferWithContext<>(wrap(buffer), buffer);
}
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 2c6355e8953..4714c4ef28d 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
@@ -19,8 +19,6 @@
package org.apache.flink.runtime.io.network.partition.consumer;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.core.memory.MemorySegment;
-import org.apache.flink.core.memory.MemorySegmentFactory;
import org.apache.flink.metrics.Counter;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
@@ -29,13 +27,10 @@ import org.apache.flink.runtime.event.TaskEvent;
import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
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.logger.NetworkActionsLogger;
import org.apache.flink.runtime.io.network.partition.ChannelStateHolder;
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.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -289,41 +284,13 @@ public abstract class RecoveredInputChannel extends InputChannel implements Chan
}
}
- /**
- * Requests a buffer for reading recovered state. {@code checkpointingDuringRecoveryEnabled} is
- * threaded from the job configuration by the caller now that the gate-level recovery flags are
- * gone; when set, the allocation may fall back to unpooled heap buffers.
- *
- * <p>FLINK-38544 transitional: removed when the spilling backend lands (together with the heap
- * fallback below, which disk spilling supersedes).
- */
- public Buffer requestBufferBlocking(boolean checkpointingDuringRecoveryEnabled)
- throws InterruptedException, IOException {
+ public Buffer requestBufferBlocking() throws InterruptedException, IOException {
// not in setup to avoid assigning buffers unnecessarily if there is no state
if (!exclusiveBuffersAssigned) {
bufferManager.requestExclusiveBuffers(networkBuffersPerChannel);
exclusiveBuffersAssigned = true;
}
- if (!checkpointingDuringRecoveryEnabled) {
- // When checkpoint-during-recovery is not enabled, the original blocking allocation
- // is used as-is — no heap buffer fallback, no behavior change from the legacy path.
- return bufferManager.requestBufferBlocking();
- }
- // Use heap buffer fallback to avoid deadlock during filtering recovery: the filtering
- // thread first requests buffers to read state (pre-filter), then requests more buffers
- // to write filtered output (post-filter). If pre-filter buffers exhaust the pool,
- // post-filter allocation blocks, stalling the thread so pre-filter buffers can never
- // be consumed and released — the thread deadlocks itself. Heap buffers bypass the pool
- // so post-filter writes always proceed. Both call sites (getBuffer and filterAndRewrite)
- // go through this method, so the fallback applies uniformly.
- // TODO: replace heap fallback with disk spilling to bound memory usage in FLINK-38544.
- Buffer buffer = bufferManager.requestBuffer();
- if (buffer != null) {
- return buffer;
- }
- MemorySegment memorySegment =
- MemorySegmentFactory.allocateUnpooledSegment(MemoryManager.DEFAULT_PAGE_SIZE);
- return new NetworkBuffer(memorySegment, FreeingBufferRecycler.INSTANCE);
+ return bufferManager.requestBufferBlocking();
}
@Override
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 c4fee251a5f..6da822a86ea 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
@@ -166,7 +166,7 @@ class RecoveredInputChannelTest {
RecoveredInputChannel channel = buildPooledChannel(pool, totalSegments);
for (int i = 0; i < totalSegments; i++) {
- channel.requestBufferBlocking(false);
+ channel.requestBufferBlocking();
}
CountDownLatch entered = new CountDownLatch(1);
@@ -176,7 +176,7 @@ class RecoveredInputChannelTest {
() -> {
try {
entered.countDown();
- result.set(channel.requestBufferBlocking(false));
+ result.set(channel.requestBufferBlocking());
} catch (Exception ignored) {
// Thread will be interrupted at teardown.
}
@@ -198,10 +198,8 @@ class RecoveredInputChannelTest {
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);
+ Buffer filterOnBuf = buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking();
+ Buffer filterOffBuf = buildPooledChannel(pool, exclusivePerChannel).requestBufferBlocking();
// Both must come from the pool — the BufferManager-owned recycler, not the
// FreeingBufferRecycler the heap-fallback used.
(flink) 03/04: [FLINK-40080] Switch flag-on recovery to fetch->drain; retire the in-memory backend
Posted by fa...@apache.org.
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 {
(flink) 01/04: [FLINK-40080][checkpoint] ChannelStateWriter#addInputDataFromSpill: replay spilled segments into the checkpoint
Posted by fa...@apache.org.
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 ddeb17e52ce13cecdc3e486971750e1807b7cbd1
Author: Rui Fan <19...@gmail.com>
AuthorDate: Mon Jul 6 02:14:11 2026 +0200
[FLINK-40080][checkpoint] ChannelStateWriter#addInputDataFromSpill: replay spilled segments into the checkpoint
ChannelStateWriter gains addInputDataFromSpill(long checkpointId,
FetchedChannelStateReader reader); the NO_OP implementation closes the
reader. ChannelStateWriterImpl enqueues
ChannelStateWriteRequest.replayInputDataFromSpill (a
CheckpointInProgressRequest whose cancel action closes the reader).
ChannelStateCheckpointWriter.writeInputFromSpill loops nextSegment(),
writes each bodyStream() via the new
ChannelStateSerializer.writeData(DataOutputStream, InputStream, int)
overload (length prefix + transferTo, fail-loud on short read), records
offset/size into pendingResult.getInputChannelOffsets() demuxed by
channelInfo, and closes the reader in a finally block. Failures
propagate via ChannelStateWriteResult.
---
.../channel/ChannelStateCheckpointWriter.java | 47 ++++++++
.../checkpoint/channel/ChannelStateSerializer.java | 14 +++
.../channel/ChannelStateWriteRequest.java | 14 +++
.../checkpoint/channel/ChannelStateWriter.java | 20 ++++
.../checkpoint/channel/ChannelStateWriterImpl.java | 7 ++
.../runtime/io/checkpointing/ChannelState.java | 18 ++-
.../io/checkpointing/InputProcessorUtil.java | 9 +-
.../SingleCheckpointBarrierHandler.java | 9 +-
.../channel/ChannelStateWriterImplTest.java | 119 ++++++++++++++++++++
.../checkpoint/channel/MockChannelStateWriter.java | 10 ++
.../runtime/io/checkpointing/ChannelStateTest.java | 124 +++++++++++++++++++--
.../checkpointing/TestBarrierHandlerFactory.java | 1 +
12 files changed, 371 insertions(+), 21 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
index 4173bb7140e..a58e924fbfc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateCheckpointWriter.java
@@ -19,6 +19,7 @@ package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.logger.NetworkActionsLogger;
import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -41,6 +42,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHANNEL_STATE_SHARED_STREAM_EXCEPTION;
@@ -161,6 +163,51 @@ class ChannelStateCheckpointWriter {
}
}
+ void writeInputFromSpill(
+ JobVertexID jobVertexID, int subtaskIndex, FetchedChannelStateReader reader) {
+ try {
+ if (isDone()) {
+ return;
+ }
+ ChannelStatePendingResult pendingResult =
+ getChannelStatePendingResult(jobVertexID, subtaskIndex);
+ runWithChecks(
+ () -> {
+ checkState(!pendingResult.isAllInputsReceived());
+ String action = "ChannelStateCheckpointWriter#writeInputFromSpill";
+ Optional<SpillSegment> next;
+ while ((next = reader.advanceAndGetNextSegment()).isPresent()) {
+ SpillSegment seg = next.get();
+ long offset = checkpointStream.getPos();
+ try (AutoCloseable ignored =
+ NetworkActionsLogger.measureIO(action, seg.channelInfo())) {
+ serializer.writeData(dataStream, seg.bodyStream(), seg.length());
+ }
+ long size = checkpointStream.getPos() - offset;
+ pendingResult
+ .getInputChannelOffsets()
+ .computeIfAbsent(
+ seg.channelInfo(), unused -> new StateContentMetaInfo())
+ .withDataAdded(offset, size);
+ NetworkActionsLogger.tracePersist(
+ action,
+ seg.length() + " bytes",
+ seg.channelInfo(),
+ checkpointId);
+ }
+ });
+ } finally {
+ try {
+ reader.close();
+ } catch (Exception e) {
+ LOG.info(
+ "Failed to close the fetched channel state reader of checkpoint {}",
+ checkpointId,
+ e);
+ }
+ }
+ }
+
void writeOutput(
JobVertexID jobVertexID, int subtaskIndex, ResultSubpartitionInfo info, Buffer buffer) {
try {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateSerializer.java
index 252d25c2e29..ec858460dd8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateSerializer.java
@@ -43,6 +43,8 @@ interface ChannelStateSerializer {
void writeData(DataOutputStream stream, Buffer... flinkBuffers) throws IOException;
+ void writeData(DataOutputStream stream, InputStream input, int length) throws IOException;
+
void readHeader(InputStream stream) throws IOException;
int readLength(InputStream stream) throws IOException;
@@ -165,6 +167,18 @@ class ChannelStateSerializerImpl implements ChannelStateSerializer {
}
}
+ @Override
+ public void writeData(DataOutputStream stream, InputStream input, int length)
+ throws IOException {
+ Preconditions.checkArgument(length >= 0, "negative state size");
+ stream.writeInt(length);
+ long copied = input.transferTo(stream);
+ if (copied != length) {
+ throw new java.io.EOFException(
+ "Unexpected EOF: expected " + length + " bytes of segment body, got " + copied);
+ }
+ }
+
private int getSize(Buffer[] buffers) {
int len = 0;
for (Buffer buffer : buffers) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
index abef241c325..d1913df0416 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriteRequest.java
@@ -258,6 +258,20 @@ abstract class ChannelStateWriteRequest {
return new CheckpointAbortRequest(jobVertexID, subtaskIndex, checkpointId, cause);
}
+ static ChannelStateWriteRequest replayInputDataFromSpill(
+ JobVertexID jobVertexID,
+ int subtaskIndex,
+ long checkpointId,
+ FetchedChannelStateReader reader) {
+ return new CheckpointInProgressRequest(
+ "writeInputFromSpill",
+ jobVertexID,
+ subtaskIndex,
+ checkpointId,
+ writer -> writer.writeInputFromSpill(jobVertexID, subtaskIndex, reader),
+ throwable -> reader.close());
+ }
+
static ChannelStateWriteRequest registerSubtask(JobVertexID jobVertexID, int subtaskIndex) {
return new SubtaskRegisterRequest(jobVertexID, subtaskIndex);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
index 6fee1402036..d10d88ea800 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriter.java
@@ -25,6 +25,9 @@ import org.apache.flink.runtime.state.InputChannelStateHandle;
import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
import org.apache.flink.util.CloseableIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.Closeable;
import java.util.Collection;
import java.util.Collections;
@@ -190,10 +193,15 @@ public interface ChannelStateWriter extends Closeable {
ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId)
throws IllegalArgumentException;
+ /** Records input-channel state from a spill file and takes ownership of {@code reader}. */
+ void addInputDataFromSpill(long checkpointId, FetchedChannelStateReader reader);
+
ChannelStateWriter NO_OP = new NoOpChannelStateWriter();
/** No-op implementation of {@link ChannelStateWriter}. */
class NoOpChannelStateWriter implements ChannelStateWriter {
+ private static final Logger LOG = LoggerFactory.getLogger(NoOpChannelStateWriter.class);
+
@Override
public void start(long checkpointId, CheckpointOptions checkpointOptions) {}
@@ -231,6 +239,18 @@ public interface ChannelStateWriter extends Closeable {
CompletableFuture.completedFuture(Collections.emptyList()));
}
+ @Override
+ public void addInputDataFromSpill(long checkpointId, FetchedChannelStateReader reader) {
+ try {
+ reader.close();
+ } catch (Exception e) {
+ LOG.info(
+ "Failed to close the fetched channel state reader of checkpoint {}",
+ checkpointId,
+ e);
+ }
+ }
+
@Override
public void close() {}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
index 40d7ddffd1e..21db97355db 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImpl.java
@@ -42,6 +42,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteRequest.completeInput;
import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteRequest.completeOutput;
+import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteRequest.replayInputDataFromSpill;
import static org.apache.flink.runtime.checkpoint.channel.ChannelStateWriteRequest.write;
/**
@@ -235,6 +236,12 @@ public class ChannelStateWriterImpl implements ChannelStateWriter {
enqueue(completeOutput(jobVertexID, subtaskIndex, checkpointId), false);
}
+ @Override
+ public void addInputDataFromSpill(long checkpointId, FetchedChannelStateReader reader) {
+ LOG.debug("{} replaying input data from spill, checkpoint {}", taskName, checkpointId);
+ enqueue(replayInputDataFromSpill(jobVertexID, subtaskIndex, checkpointId, reader), false);
+ }
+
@Override
public void abort(long checkpointId, Throwable cause, boolean cleanup) {
LOG.debug("{} aborting, checkpoint {}", taskName, checkpointId);
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java
index b0dfb4291a1..31288059fd3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelState.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.runtime.io.checkpointing;
import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateSnapshot;
import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger;
@@ -54,14 +55,19 @@ final class ChannelState {
private final RecoveryCheckpointTrigger recoveryCheckpointTrigger;
+ private final ChannelStateWriter channelStateWriter;
+
public ChannelState(CheckpointableInput[] inputs) {
- this(inputs, RecoveryCheckpointTrigger.NO_OP);
+ this(inputs, RecoveryCheckpointTrigger.NO_OP, ChannelStateWriter.NO_OP);
}
public ChannelState(
- CheckpointableInput[] inputs, RecoveryCheckpointTrigger recoveryCheckpointTrigger) {
+ CheckpointableInput[] inputs,
+ RecoveryCheckpointTrigger recoveryCheckpointTrigger,
+ ChannelStateWriter channelStateWriter) {
this.inputs = inputs;
this.recoveryCheckpointTrigger = checkNotNull(recoveryCheckpointTrigger);
+ this.channelStateWriter = checkNotNull(channelStateWriter);
}
public void blockChannel(InputChannelInfo channelInfo) {
@@ -113,19 +119,19 @@ final class ChannelState {
}
/**
- * Dispatches checkpoint start: inserts recovery-checkpoint barriers into in-recovery channels
- * through the trigger, then notifies every input. (FLINK-38544 transitional: the spilling
- * backend adds a third step handing a reader opened from the snapshot to the channel-state
- * writer, instead of closing it here.)
+ * Transfers spill-snapshot ownership to the writer after all inputs observe checkpoint start.
*/
public void onCheckpointStartedForAllInputs(CheckpointBarrier barrier)
throws CheckpointException, IOException {
long cpId = barrier.getId();
+ // The snapshot is closed either way: closing is a no-op once the reader below was opened
+ // (the writer owns and closes it), and releases the grant if we never got that far.
try (FetchedChannelStateSnapshot snapshot =
recoveryCheckpointTrigger.snapshotAndInsertBarriers(cpId)) {
for (CheckpointableInput input : inputs) {
input.checkpointStarted(barrier);
}
+ channelStateWriter.addInputDataFromSpill(cpId, snapshot.reader());
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java
index 39f85fe08b2..e299399e993 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/InputProcessorUtil.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.CheckpointingMode;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger;
import org.apache.flink.runtime.io.network.partition.consumer.CheckpointableInput;
import org.apache.flink.runtime.io.network.partition.consumer.IndexedInputGate;
@@ -123,6 +124,7 @@ public class InputProcessorUtil {
Clock clock = SystemClock.getInstance();
CheckpointingMode checkpointingMode = CheckpointingOptions.getCheckpointingMode(jobConf);
+ ChannelStateWriter channelStateWriter = checkpointCoordinator.getChannelStateWriter();
switch (checkpointingMode) {
case EXACTLY_ONCE:
int numberOfChannels =
@@ -141,7 +143,8 @@ public class InputProcessorUtil {
inputs,
clock,
numberOfChannels,
- recoveryCheckpointTrigger);
+ recoveryCheckpointTrigger,
+ channelStateWriter);
case AT_LEAST_ONCE:
if (CheckpointingOptions.isUnalignedCheckpointEnabled(jobConf)) {
throw new IllegalStateException(
@@ -175,7 +178,8 @@ public class InputProcessorUtil {
CheckpointableInput[] inputs,
Clock clock,
int numberOfChannels,
- RecoveryCheckpointTrigger recoveryCheckpointTrigger) {
+ RecoveryCheckpointTrigger recoveryCheckpointTrigger,
+ ChannelStateWriter channelStateWriter) {
boolean enableCheckpointAfterTasksFinished =
config.getConfiguration()
.get(CheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH);
@@ -189,6 +193,7 @@ public class InputProcessorUtil {
BarrierAlignmentUtil.createRegisterTimerCallback(mailboxExecutor, timerService),
enableCheckpointAfterTasksFinished,
recoveryCheckpointTrigger,
+ channelStateWriter,
inputs);
} else {
return SingleCheckpointBarrierHandler.aligned(
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
index 8594fba540f..547941c2447 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/io/checkpointing/SingleCheckpointBarrierHandler.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointFailureReason;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger;
import org.apache.flink.runtime.io.network.api.CancelCheckpointMarker;
@@ -121,6 +122,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
},
enableCheckpointsAfterTasksFinish,
RecoveryCheckpointTrigger.NO_OP,
+ ChannelStateWriter.NO_OP,
inputs);
}
@@ -133,6 +135,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
DelayableTimer registerTimer,
boolean enableCheckpointAfterTasksFinished,
RecoveryCheckpointTrigger recoveryCheckpointTrigger,
+ ChannelStateWriter channelStateWriter,
CheckpointableInput... inputs) {
return new SingleCheckpointBarrierHandler(
taskName,
@@ -141,7 +144,8 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
clock,
numOpenChannels,
new AlternatingWaitingForFirstBarrierUnaligned(
- false, new ChannelState(inputs, recoveryCheckpointTrigger)),
+ false,
+ new ChannelState(inputs, recoveryCheckpointTrigger, channelStateWriter)),
false,
registerTimer,
inputs,
@@ -178,6 +182,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
DelayableTimer registerTimer,
boolean enableCheckpointAfterTasksFinished,
RecoveryCheckpointTrigger recoveryCheckpointTrigger,
+ ChannelStateWriter channelStateWriter,
CheckpointableInput... inputs) {
return new SingleCheckpointBarrierHandler(
taskName,
@@ -186,7 +191,7 @@ public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
clock,
numOpenChannels,
new AlternatingWaitingForFirstBarrier(
- new ChannelState(inputs, recoveryCheckpointTrigger)),
+ new ChannelState(inputs, recoveryCheckpointTrigger, channelStateWriter)),
true,
registerTimer,
inputs,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
index fb931946bbb..50ec1cc6afb 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/ChannelStateWriterImplTest.java
@@ -30,11 +30,14 @@ import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.util.function.BiConsumerWithException;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.util.ArrayDeque;
+import java.util.Collections;
import java.util.Deque;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import static org.apache.flink.util.CloseableIterator.ofElements;
@@ -51,6 +54,8 @@ class ChannelStateWriterImplTest {
private static final CheckpointStorage CHECKPOINT_STORAGE = new JobManagerCheckpointStorage();
+ @TempDir private java.nio.file.Path tempDir;
+
@Test
void testAddEventBuffer() throws Exception {
@@ -371,10 +376,124 @@ class ChannelStateWriterImplTest {
writer.finishInput(CHECKPOINT_ID);
writer.finishOutput(CHECKPOINT_ID);
}
+
+ @Test
+ void testAddInputDataFromSpillAsyncDemux() throws Exception {
+ SyncChannelStateWriteRequestExecutor worker =
+ new SyncChannelStateWriteRequestExecutor(JOB_ID);
+ try (ChannelStateWriterImpl writer = newWriter(worker)) {
+ worker.registerSubtask(JOB_VERTEX_ID, SUBTASK_INDEX);
+ writer.start(CHECKPOINT_ID, CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ InputChannelInfo c0 = new InputChannelInfo(0, 0);
+ InputChannelInfo c1 = new InputChannelInfo(0, 1);
+
+ FetchedChannelState state;
+ try (TestSpillWriter spillWriter = new TestSpillWriter(tempDir)) {
+ spillWriter.writeRecord(c0, new byte[] {1, 2, 3}, 3);
+ spillWriter.writeRecord(c1, new byte[] {4, 5}, 2);
+ spillWriter.writeRecord(c0, new byte[] {6}, 1);
+ state = spillWriter.getChannelState();
+ }
+ FetchedChannelStateReader reader = state.reader();
+ // Drop the handoff grant; the reader now holds the only outstanding grant.
+ state.release();
+
+ writer.addInputDataFromSpill(CHECKPOINT_ID, reader);
+ // Request is queued but not yet processed — state must still be alive.
+ assertThat(state.isClosed()).isFalse();
+
+ worker.processAllRequests();
+ // After processing, the reader is closed by the request, releasing the last grant.
+ assertThat(state.isClosed()).isTrue();
+ }
+ }
+
+ @Test
+ void testAddInputDataFromSpillEmptySnapshotStillSubmitted() throws Exception {
+ // Empty readers are no longer short-circuited; they are submitted to the writer thread.
+ QueueCountingExecutor worker = new QueueCountingExecutor();
+ try (ChannelStateWriterImpl writer =
+ new ChannelStateWriterImpl(
+ JOB_VERTEX_ID,
+ TASK_NAME,
+ SUBTASK_INDEX,
+ new ConcurrentHashMap<>(),
+ worker,
+ 5)) {
+ worker.registerSubtask(JOB_VERTEX_ID, SUBTASK_INDEX);
+ writer.start(CHECKPOINT_ID, CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ int submittedBefore = worker.submitCount.get();
+ FetchedChannelState emptyState = new FetchedChannelState(Collections.emptyList());
+ FetchedChannelStateReader emptyReader = emptyState.reader();
+ emptyState.release();
+
+ writer.addInputDataFromSpill(CHECKPOINT_ID, emptyReader);
+
+ assertThat(worker.submitCount.get())
+ .as("empty reader must still be submitted to the writer thread")
+ .isGreaterThan(submittedBefore);
+ }
+ }
+
+ @Test
+ void testAddInputDataFromSpillSegmentsClosedOnSuccess() throws Exception {
+ SyncChannelStateWriteRequestExecutor worker =
+ new SyncChannelStateWriteRequestExecutor(JOB_ID);
+ try (ChannelStateWriterImpl writer = newWriter(worker)) {
+ worker.registerSubtask(JOB_VERTEX_ID, SUBTASK_INDEX);
+ writer.start(CHECKPOINT_ID, CheckpointOptions.forCheckpointWithDefaultLocation());
+
+ FetchedChannelState state;
+ try (TestSpillWriter spillWriter = new TestSpillWriter(tempDir)) {
+ spillWriter.writeRecord(new InputChannelInfo(0, 0), new byte[] {1}, 1);
+ state = spillWriter.getChannelState();
+ }
+ FetchedChannelStateReader reader = state.reader();
+ state.release();
+
+ writer.addInputDataFromSpill(CHECKPOINT_ID, reader);
+ worker.processAllRequests();
+
+ // After processing, the last grant is released and the state is cleaned up.
+ assertThat(state.isClosed()).isTrue();
+ }
+ }
+
+ private ChannelStateWriterImpl newWriter(SyncChannelStateWriteRequestExecutor worker) {
+ return new ChannelStateWriterImpl(
+ JOB_VERTEX_ID, TASK_NAME, SUBTASK_INDEX, new ConcurrentHashMap<>(), worker, 5);
+ }
}
class TestException extends RuntimeException {}
+/** Counts submissions without processing them, for the empty-snapshot submission test. */
+class QueueCountingExecutor implements ChannelStateWriteRequestExecutor {
+
+ final AtomicInteger submitCount = new AtomicInteger(0);
+
+ @Override
+ public void submit(ChannelStateWriteRequest e) {
+ submitCount.incrementAndGet();
+ }
+
+ @Override
+ public void submitPriority(ChannelStateWriteRequest e) {
+ submitCount.incrementAndGet();
+ }
+
+ @Override
+ public void start() throws IllegalStateException {}
+
+ @Override
+ public void registerSubtask(JobVertexID jobVertexID, int subtaskIndex) {}
+
+ @Override
+ public void releaseSubtask(JobVertexID jobVertexID, int subtaskIndex) {}
+}
+
class SyncChannelStateWriteRequestExecutor implements ChannelStateWriteRequestExecutor {
private final ChannelStateWriteRequestDispatcher requestProcessor;
private final Deque<ChannelStateWriteRequest> deque;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
index c77208f3ff7..bdff6d44718 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/MockChannelStateWriter.java
@@ -74,6 +74,16 @@ public class MockChannelStateWriter implements ChannelStateWriter {
}
}
+ @Override
+ public void addInputDataFromSpill(long checkpointId, FetchedChannelStateReader reader) {
+ checkCheckpointId(checkpointId);
+ try {
+ reader.close();
+ } catch (Exception e) {
+ rethrow(e);
+ }
+ }
+
@Override
public void addOutputData(
long checkpointId, ResultSubpartitionInfo info, int startSeqNum, Buffer... data) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java
index f6dc7f7f69b..0307a92f1af 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/ChannelStateTest.java
@@ -21,29 +21,34 @@ package org.apache.flink.streaming.runtime.io.checkpointing;
import org.apache.flink.runtime.checkpoint.CheckpointException;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
import org.apache.flink.runtime.checkpoint.channel.FetchedChannelState;
+import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader;
import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateSnapshot;
import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
import org.apache.flink.runtime.checkpoint.channel.RecoveryCheckpointTrigger;
+import org.apache.flink.runtime.checkpoint.channel.ResultSubpartitionInfo;
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.partition.consumer.CheckpointableInput;
import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
+import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import static org.assertj.core.api.Assertions.assertThat;
/**
* Verifies the {@link ChannelState#onCheckpointStartedForAllInputs} dispatcher: call ordering and
* feature-off no-op routing through the {@link RecoveryCheckpointTrigger#NO_OP} singleton.
- *
- * <p>FLINK-38544 transitional: this covers the 2-step dispatch (trigger, then per-input
- * notification); the spilling backend adds a third step handing the trigger's snapshot reader to
- * the channel-state writer and completes this test to cover all three steps.
*/
class ChannelStateTest {
@@ -52,11 +57,15 @@ class ChannelStateTest {
@Test
void testStepOrderingFeatureOn() throws Exception {
List<String> trace = new ArrayList<>();
- RecordingTrigger trigger = new RecordingTrigger(trace);
+ // An empty reader is sufficient to verify ordering.
+ FetchedChannelStateSnapshot snap = FetchedChannelState.emptySnapshot();
+ RecordingTrigger trigger = new RecordingTrigger(trace, snap);
+ RecordingWriter writer = new RecordingWriter(trace);
CheckpointableInput input1 = new RecordingInput(trace, "in1");
CheckpointableInput input2 = new RecordingInput(trace, "in2");
- ChannelState state = new ChannelState(new CheckpointableInput[] {input1, input2}, trigger);
+ ChannelState state =
+ new ChannelState(new CheckpointableInput[] {input1, input2}, trigger, writer);
CheckpointBarrier barrier = newUnalignedBarrier();
state.onCheckpointStartedForAllInputs(barrier);
@@ -65,21 +74,49 @@ class ChannelStateTest {
.containsExactly(
"trigger.snapshotAndInsertBarriers:" + CHECKPOINT_ID,
"in1.checkpointStarted:" + CHECKPOINT_ID,
- "in2.checkpointStarted:" + CHECKPOINT_ID);
+ "in2.checkpointStarted:" + CHECKPOINT_ID,
+ "writer.addInputDataFromSpill:" + CHECKPOINT_ID);
}
@Test
void testStepOrderingFeatureOff() throws Exception {
List<String> trace = new ArrayList<>();
+ RecordingWriter writer = new RecordingWriter(trace);
CheckpointableInput input = new RecordingInput(trace, "in1");
ChannelState state =
new ChannelState(
- new CheckpointableInput[] {input}, RecoveryCheckpointTrigger.NO_OP);
+ new CheckpointableInput[] {input}, RecoveryCheckpointTrigger.NO_OP, writer);
+
+ state.onCheckpointStartedForAllInputs(newUnalignedBarrier());
+
+ assertThat(trace)
+ .containsExactly(
+ "in1.checkpointStarted:" + CHECKPOINT_ID,
+ "writer.addInputDataFromSpill:" + CHECKPOINT_ID);
+ assertThat(writer.lastSnapshotWasEmpty.get()).isTrue();
+ }
+
+ @Test
+ void testEmptySnapshotStillSubmitted() throws Exception {
+ // Empty readers (no spill files) are no longer short-circuited; they still reach
+ // addInputDataFromSpill on the writer thread.
+ List<String> trace = new ArrayList<>();
+ FetchedChannelStateSnapshot emptySnap = FetchedChannelState.emptySnapshot();
+ RecordingTrigger trigger = new RecordingTrigger(trace, emptySnap);
+ RecordingWriter writer = new RecordingWriter(trace);
+
+ ChannelState state =
+ new ChannelState(
+ new CheckpointableInput[] {new RecordingInput(trace, "in1")},
+ trigger,
+ writer);
state.onCheckpointStartedForAllInputs(newUnalignedBarrier());
- assertThat(trace).containsExactly("in1.checkpointStarted:" + CHECKPOINT_ID);
+ // Empty reader must still reach the writer (no inline short-circuit).
+ assertThat(writer.addInputDataFromSpillCalls.get()).isEqualTo(1);
+ assertThat(writer.lastSnapshotWasEmpty.get()).isTrue();
}
private static CheckpointBarrier newUnalignedBarrier() {
@@ -93,18 +130,83 @@ class ChannelStateTest {
private static final class RecordingTrigger implements RecoveryCheckpointTrigger {
private final List<String> trace;
+ private final FetchedChannelStateSnapshot snapshot;
- RecordingTrigger(List<String> trace) {
+ RecordingTrigger(List<String> trace, FetchedChannelStateSnapshot snapshot) {
this.trace = trace;
+ this.snapshot = snapshot;
}
@Override
public FetchedChannelStateSnapshot snapshotAndInsertBarriers(long checkpointId) {
trace.add("trigger.snapshotAndInsertBarriers:" + checkpointId);
- return FetchedChannelState.emptySnapshot();
+ return snapshot;
}
}
+ private static final class RecordingWriter implements ChannelStateWriter {
+ private final List<String> trace;
+ final AtomicBoolean lastSnapshotWasEmpty = new AtomicBoolean(false);
+ final AtomicLong lastCpId = new AtomicLong(-1L);
+ final AtomicInteger addInputDataFromSpillCalls = new AtomicInteger(0);
+
+ RecordingWriter(List<String> trace) {
+ this.trace = trace;
+ }
+
+ @Override
+ public void start(long checkpointId, CheckpointOptions checkpointOptions) {}
+
+ @Override
+ public void addInputData(
+ long checkpointId,
+ InputChannelInfo info,
+ int startSeqNum,
+ CloseableIterator<Buffer> data) {}
+
+ @Override
+ public void addOutputData(
+ long checkpointId, ResultSubpartitionInfo info, int startSeqNum, Buffer... data) {}
+
+ @Override
+ public void addOutputDataFuture(
+ long checkpointId,
+ ResultSubpartitionInfo info,
+ int startSeqNum,
+ CompletableFuture<List<Buffer>> data) {}
+
+ @Override
+ public void finishInput(long checkpointId) {}
+
+ @Override
+ public void finishOutput(long checkpointId) {}
+
+ @Override
+ public void abort(long checkpointId, Throwable cause, boolean cleanup) {}
+
+ @Override
+ public ChannelStateWriteResult getAndRemoveWriteResult(long checkpointId) {
+ return ChannelStateWriteResult.EMPTY;
+ }
+
+ @Override
+ public void addInputDataFromSpill(long checkpointId, FetchedChannelStateReader reader) {
+ trace.add("writer.addInputDataFromSpill:" + checkpointId);
+ lastCpId.set(checkpointId);
+ addInputDataFromSpillCalls.incrementAndGet();
+ try {
+ // Peek whether the reader has any segments by attempting the first advance.
+ // The first nextSegment() call is exempt from the "previous body consumed" rule.
+ lastSnapshotWasEmpty.set(reader.advanceAndGetNextSegment().isEmpty());
+ reader.close();
+ } catch (Exception ignored) {
+ }
+ }
+
+ @Override
+ public void close() {}
+ }
+
private static final class RecordingInput implements CheckpointableInput {
private final List<String> trace;
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java
index dca71e35f6b..8c281c04a6b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/TestBarrierHandlerFactory.java
@@ -74,6 +74,7 @@ public class TestBarrierHandlerFactory {
actionRegistration,
enableCheckpointsAfterTasksFinish,
RecoveryCheckpointTrigger.NO_OP,
+ stateWriter,
inputGate);
}
}