1996fanrui commented on code in PR #28661: URL: https://github.com/apache/flink/pull/28661#discussion_r3719813640
########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReader.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.annotation.Internal; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Collections; +import java.util.Optional; + +/** + * Forward reader over a {@link FetchedChannelState}'s spill files. This is our own segment reader, + * on purpose <em>not</em> a Java {@link java.util.Iterator}: our access pattern ("a body must be + * fully read before the next segment", "body ownership is handed to the consumer", "consume and + * commit are separate steps") does not fit the {@code hasNext/next} contract. + * + * <p>This interface is the contract callers depend on; {@link FetchedChannelStateReaderImpl} holds + * the implementation (the live file stream, the two progress positions, the bounded body view). + * + * <p>Reading is strictly sequential: a reader is positioned once (offset 0 for the root reader, or + * the committed position for a {@link #snapshot()}), then consumes forward only via {@link + * #nextSegment()}. It never seeks backward and never re-positions mid-iteration. + * + * <p>The drain thread reads the root reader front to back and commits via {@link Review Comment: It was just informal terminology — replaced "root reader" everywhere with "drain reader" and defined it in the interface javadoc. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; Review Comment: Right, removed the field in favor of `snapshot.channelState()`. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReader.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.annotation.Internal; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Collections; +import java.util.Optional; + +/** + * Forward reader over a {@link FetchedChannelState}'s spill files. This is our own segment reader, + * on purpose <em>not</em> a Java {@link java.util.Iterator}: our access pattern ("a body must be + * fully read before the next segment", "body ownership is handed to the consumer", "consume and + * commit are separate steps") does not fit the {@code hasNext/next} contract. + * + * <p>This interface is the contract callers depend on; {@link FetchedChannelStateReaderImpl} holds + * the implementation (the live file stream, the two progress positions, the bounded body view). + * + * <p>Reading is strictly sequential: a reader is positioned once (offset 0 for the root reader, or + * the committed position for a {@link #snapshot()}), then consumes forward only via {@link + * #nextSegment()}. It never seeks backward and never re-positions mid-iteration. + * + * <p>The drain thread reads the root reader front to back and commits via {@link + * SpillSegment#commit()}; each checkpoint derives a fresh {@link #snapshot()} that resumes from the + * committed position. {@link #snapshot()} and {@link SpillSegment#commit()} must be called under + * the drainer lock; disk reads happen outside it. + */ +@Internal +public interface FetchedChannelStateReader extends Closeable { + + /** + * Advances to the next segment and returns it, or {@link Optional#empty()} when no segment + * remains. Advancing and probing are one step; there is no separate {@code hasNext}. + * + * <p>Entry rule (the first call is exempt): the previous segment's body must be fully read, + * otherwise this is a contract violation and fails loud (no skip-ahead). + */ + Optional<SpillSegment> nextSegment(); + + /** + * Derives an independent resume point starting from the committed position. The snapshot holds + * its own {@link FetchedChannelState} lifecycle grant; the caller must open a reader from it + * via {@link FetchedChannelStateSnapshot#reader()} and close that reader when done. + * + * <p>Must be called under the drainer lock so that the copied position reflects the latest + * committed state. + * + * @return a snapshot capturing the current committed position; caller must open and close a + * reader from it + */ + FetchedChannelStateSnapshot snapshot(); Review Comment: Not intentional — made `FetchedChannelStateSnapshot` `@Internal public` with a public `reader()`. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; + private final List<Path> files; + + /** Live read position; {@code readOffset} is where the open stream physically sits. */ + private final Position current; + + /** Delivered boundary; {@link SpillSegment#commit()} advances it from {@link #current}. */ + private final Position committed; + + /** Open stream over {@code current.fileIndex}, or {@code null} before the first read. */ + @Nullable private InputStream fileStream; + + /** Size of the file currently open. */ + private long currentFileSize; + + /** Body view of the segment returned by the last {@link #nextSegment()}, or {@code null}. */ + @Nullable private BoundedSegmentStream currentBody; + + private boolean positioned; + private boolean closed; + + FetchedChannelStateReaderImpl(FetchedChannelStateSnapshot snapshot) { + this.snapshot = snapshot; + this.channelState = snapshot.channelState(); + this.files = channelState.files(); + // Must copy the position so that this reader's commits do not mutate the snapshot's state. + this.committed = snapshot.position().copy(); + this.current = committed.copy(); + } + + @Override + public Optional<SpillSegment> nextSegment() { + checkState(!closed, "FetchedChannelStateReader is closed"); + checkState( + currentBody == null || currentBody.remaining() == 0, + "Previous segment body not fully consumed before advancing: %s bytes left", + currentBody == null ? 0 : currentBody.remaining()); + try { + if (!positioned) { + positioned = true; + return firstSegment(); + } + return followingSegment(); + } catch (IOException e) { + throw new RuntimeException("Failed to read segment", e); + } + } + + /** + * First positioning — the only path that may skip bytes. Opens the file at the committed header + * offset and reads the header. A snapshot may resume in the middle of a segment: the committed + * {@code readOffset} says how many body bytes were already delivered, and that prefix is + * skipped so the returned body starts at the not-yet-delivered remainder. If the segment was + * already fully delivered (prefix == whole body), it is exhausted here and we move on to the + * next one. + */ + private Optional<SpillSegment> firstSegment() throws IOException { + // The committed read offset may sit mid-body (after a partial commit), but the header lives + // at segmentStartOffset. Capture how much was already delivered, then rewind the live read + // offset to the header so we open the file there and read the header, not mid-body. + int deliveredPrefix = (int) current.deliveredBodyBytes(); + current.rewindToSegmentStart(); + + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + checkState( + deliveredPrefix <= header.bufferLength, + "Delivered offset %s exceeds segment length %s", + deliveredPrefix, + header.bufferLength); + + if (deliveredPrefix == header.bufferLength) { + // This segment was already fully delivered before the snapshot; nothing remains in it. + // Skip its whole body to reach the next segment's header, then take the steady path. + skipBody(header.bufferLength); + return followingSegment(); + } + + // Discard the already-delivered prefix (the one and only skip in this class), then hand out + // the remainder. alreadyDelivered is carried so commit() records the boundary from the + // head. + skipBody(deliveredPrefix); + currentBody = + new BoundedSegmentStream(header.bufferLength - deliveredPrefix, deliveredPrefix); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + /** + * Steady-state path — no skipping. The previous body was read to its end, so the stream sits + * exactly on this segment's header (or at the current file's end, in which case we roll to the + * next file). Reads the header and returns the whole-body view. + */ + private Optional<SpillSegment> followingSegment() throws IOException { + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + currentBody = new BoundedSegmentStream(header.bufferLength); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + @Override + public FetchedChannelStateSnapshot snapshot() { + checkState(!closed, "FetchedChannelStateReader is closed"); + return new FetchedChannelStateSnapshot(channelState, committed.copy()); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + closed = true; + try { + closeFileStream(); + } finally { + snapshot.release(); + } + } + + // ------------------------------------------------------------------------------------------- + // Sequential IO over the spill files; all of it advances current.readOffset / current.fileIndex + // ------------------------------------------------------------------------------------------- + + /** + * Ensures a file is open with the stream positioned at {@code current}'s read offset, ready to + * read this segment's header. Rolls to the next file when the current one is exhausted. Returns + * false when no segment remains. + * + * <p>{@code current.segmentStartOffset} is set to where the header begins, so a later {@link + * SpillSegment#commit()} records the right segment for the snapshot to resume from. + */ + private boolean openCurrentFile() throws IOException { + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + // Current file fully read: move to the next file's first segment. + closeFileStream(); + current.rollToNextFile(); + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + return false; + } + + /** Reads the 12-byte header at the current read offset; advances past it. */ + private SegmentHeader readHeaderAtCurrent() throws IOException { + byte[] headerBytes = new byte[SEGMENT_HEADER_BYTES]; + readFully(headerBytes); + DataInputStream h = new DataInputStream(new ByteArrayInputStream(headerBytes)); + int gateIdx = h.readInt(); + int channelIdx = h.readInt(); + int bufferLength = h.readInt(); + checkState(bufferLength >= 0, "negative segment length: %s", bufferLength); + return new SegmentHeader(new InputChannelInfo(gateIdx, channelIdx), bufferLength); + } + + /** + * Ensures the file at {@code current.fileIndex} is open with the stream positioned at {@code + * current.readOffset}. If a stream is already open it is left as-is: sequential reading + * guarantees it is already there. + */ + private void openFileAndSeek() throws IOException { + if (fileStream != null) { + return; + } + Path path = files.get(current.fileIndex); + currentFileSize = Files.size(path); + InputStream in = Files.newInputStream(path); + try { Review Comment: Not intentional — wrapped it in a `BufferedInputStream` ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImpl.java: ########## @@ -110,7 +111,7 @@ public Optional<FetchedChannelState> readInputData( // only signals "there is state to recover". The spilling backend returns a real, // file-backed container here. Review Comment: Compressed the first paragraph to the non-obvious part (why the condition is "readAny" rather than "a filter ran") and added the TODO about returning `getProducedChannelState()` once the `Spilling*` handlers are selected. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java: ########## @@ -214,9 +234,368 @@ public void recover( } } +/** + * Intermediate abstract base for the two spilling variants. Owns the on-disk spill format end to + * end: a single reusable {@link DataOutputSerializer} accumulates one channel's segment, the + * segment header is backfilled with the body length at seal time, and sealed segments are flushed + * to the current file stream with 64 MB-bounded rotation. + * + * <h3>Disk format</h3> + * + * <pre> + * [ 4B BE int: gate idx ] segment header: written once per channel segment + * [ 4B BE int: channel idx ] + * [ 4B BE int: buffer length ] segment body byte count (backfilled at segment seal) + * [ 4B BE int: record length ] repeated for every record in this segment + * [ N bytes: serialized record ] + * [ 4B BE int: gate idx ] next segment header (channel switch or post-rotation) + * ... + * </pre> + * + * <p>The body byte count is only known after the whole segment is written, so each segment is first + * accumulated in {@link #segmentSerializer} (header written at open with a zero placeholder) and + * {@link DataOutputSerializer#writeIntUnsafe} backfills the length at seal. A segment is one + * uninterrupted run of records for a single channel; file rotation happens only after a segment is + * fully sealed, so a segment never crosses a file boundary. + */ +abstract class AbstractSpillingHandler extends AbstractInputChannelRecoveredStateHandler { + + /** Byte offset of the {@code bufferLength} field within a segment's header. */ + static final int BUFFER_LENGTH_HEADER_OFFSET = 2 * Integer.BYTES; + + /** Total size of the segment header in bytes: gateIdx + channelIdx + bufferLength. */ + static final int SEGMENT_HEADER_BYTES = 3 * Integer.BYTES; + + final String[] spillTmpDirectories; + + public static final long DEFAULT_SPILL_FILE_SIZE_BYTES = 64L * 1024 * 1024; + + /** Soft per-file size bound that triggers rotation between segments. */ + private final long maxFileSizeBytes; + + /** + * Accumulates the current segment: the header followed by the body, which is either + * length-prefixed filtered records or verbatim pass-through bytes, depending on the subclass. + * Reused across segments via {@code clear()}. + */ + private final DataOutputSerializer segmentSerializer = new DataOutputSerializer(256); + + /** + * Spill files written so far, in order. The {@link FetchedChannelState} handoff is built from + * this list once writing is sealed; an empty list means the handler never spilled any bytes, so + * it produces no state. + */ + private final List<Path> files = new ArrayList<>(); + + /** + * Unique directory for this handler's spill files; created lazily when the first file opens. + */ + private final Path baseDir; + + /** + * Output stream to the current spill file; tracks the bytes written so far via {@link + * OffsetAwareOutputStream#getLength()} to decide when to rotate. Null before the first segment + * is flushed. + */ + @Nullable private OffsetAwareOutputStream currentStream; + + /** Channel whose segment is currently open; null when no segment is in progress. */ + @Nullable private InputChannelInfo currentChannel; + + @Nullable private FetchedChannelState producedChannelState; + + AbstractSpillingHandler( + InputGate[] inputGates, + InflightDataRescalingDescriptor channelMapping, + String[] spillTmpDirectories, + long maxFileSizeBytes) { + // 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); + checkArgument( + checkNotNull(spillTmpDirectories).length > 0, + "spillTmpDirectories must not be empty"); + checkArgument( + maxFileSizeBytes > 0, "maxFileSizeBytes must be positive: %s", maxFileSizeBytes); + this.spillTmpDirectories = spillTmpDirectories; + this.maxFileSizeBytes = maxFileSizeBytes; + this.baseDir = + Paths.get(spillTmpDirectories[0], "flink-channel-spill-" + UUID.randomUUID()); + } + + /** + * Opens (or switches to) the segment for {@code channelInfo} and returns its buffer for the + * caller to append the body into. The caller must not seal the segment. + */ + DataOutputSerializer segmentSerializerFor(InputChannelInfo channelInfo) throws IOException { + switchChannelIfNeeded(channelInfo); + return segmentSerializer; + } + + private void switchChannelIfNeeded(InputChannelInfo channelInfo) throws IOException { + if (channelInfo.equals(currentChannel)) { + return; + } + if (currentChannel != null) { + sealCurrentSegment(); + } + segmentSerializer.clear(); + segmentSerializer.writeInt(channelInfo.getGateIdx()); + segmentSerializer.writeInt(channelInfo.getInputChannelIdx()); + segmentSerializer.writeInt(0); // bufferLength placeholder + currentChannel = channelInfo; + } + + /** + * Backfills the body length into the segment header and flushes the whole segment to the file + * stream. Empty segments (filtered out entirely, or a zero-byte pass-through) are dropped + * without opening a file, so no empty file is created. + */ + private void sealCurrentSegment() throws IOException { + if (currentChannel == null) { + return; + } + currentChannel = null; + int totalBytes = segmentSerializer.length(); + int bodyBytes = totalBytes - SEGMENT_HEADER_BYTES; + if (bodyBytes == 0) { + return; + } + // Math.toIntExact guards against the unlikely case of a single segment > 2 GB. + segmentSerializer.writeIntUnsafe(Math.toIntExact(bodyBytes), BUFFER_LENGTH_HEADER_OFFSET); + ensureFileOpen(); + currentStream.write(segmentSerializer.getSharedBuffer(), 0, totalBytes); + } + + /** + * Ensures an output stream is ready for the next segment, rotating to a fresh file first if the + * current one reached the size bound. Rotation happens here, between sealed segments, so a + * segment is never split across files. + */ + private void ensureFileOpen() throws IOException { + if (currentStream != null && currentStream.getLength() >= maxFileSizeBytes) { + currentStream.flush(); + currentStream.close(); + currentStream = null; + } + if (currentStream != null) { + return; + } + // create the spill dir on the first file; no-op afterwards + Files.createDirectories(baseDir); + Path filePath = baseDir.resolve("spill-segment-" + files.size() + ".bin"); + currentStream = + new OffsetAwareOutputStream( + new BufferedOutputStream(new FileOutputStream(filePath.toFile())), 0L); Review Comment: Done — the file is now opened with `Files.newOutputStream(CREATE_NEW, WRITE)` to fail loudly on collisions. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReader.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.annotation.Internal; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Collections; +import java.util.Optional; + +/** + * Forward reader over a {@link FetchedChannelState}'s spill files. This is our own segment reader, Review Comment: Trimmed the javadoc down to two paragraphs; kept the name `nextSegment` since it reads more domain-specific to me than `advanceAndGet`. wdyt? ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; + private final List<Path> files; + + /** Live read position; {@code readOffset} is where the open stream physically sits. */ + private final Position current; + + /** Delivered boundary; {@link SpillSegment#commit()} advances it from {@link #current}. */ + private final Position committed; + + /** Open stream over {@code current.fileIndex}, or {@code null} before the first read. */ + @Nullable private InputStream fileStream; + + /** Size of the file currently open. */ + private long currentFileSize; + + /** Body view of the segment returned by the last {@link #nextSegment()}, or {@code null}. */ + @Nullable private BoundedSegmentStream currentBody; + + private boolean positioned; + private boolean closed; + + FetchedChannelStateReaderImpl(FetchedChannelStateSnapshot snapshot) { + this.snapshot = snapshot; + this.channelState = snapshot.channelState(); + this.files = channelState.files(); + // Must copy the position so that this reader's commits do not mutate the snapshot's state. + this.committed = snapshot.position().copy(); + this.current = committed.copy(); + } + + @Override + public Optional<SpillSegment> nextSegment() { + checkState(!closed, "FetchedChannelStateReader is closed"); + checkState( + currentBody == null || currentBody.remaining() == 0, + "Previous segment body not fully consumed before advancing: %s bytes left", + currentBody == null ? 0 : currentBody.remaining()); + try { + if (!positioned) { + positioned = true; + return firstSegment(); + } + return followingSegment(); + } catch (IOException e) { + throw new RuntimeException("Failed to read segment", e); + } + } + + /** + * First positioning — the only path that may skip bytes. Opens the file at the committed header + * offset and reads the header. A snapshot may resume in the middle of a segment: the committed + * {@code readOffset} says how many body bytes were already delivered, and that prefix is + * skipped so the returned body starts at the not-yet-delivered remainder. If the segment was + * already fully delivered (prefix == whole body), it is exhausted here and we move on to the + * next one. + */ + private Optional<SpillSegment> firstSegment() throws IOException { + // The committed read offset may sit mid-body (after a partial commit), but the header lives + // at segmentStartOffset. Capture how much was already delivered, then rewind the live read + // offset to the header so we open the file there and read the header, not mid-body. + int deliveredPrefix = (int) current.deliveredBodyBytes(); + current.rewindToSegmentStart(); + + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + checkState( + deliveredPrefix <= header.bufferLength, + "Delivered offset %s exceeds segment length %s", + deliveredPrefix, + header.bufferLength); + + if (deliveredPrefix == header.bufferLength) { + // This segment was already fully delivered before the snapshot; nothing remains in it. + // Skip its whole body to reach the next segment's header, then take the steady path. + skipBody(header.bufferLength); + return followingSegment(); + } + + // Discard the already-delivered prefix (the one and only skip in this class), then hand out + // the remainder. alreadyDelivered is carried so commit() records the boundary from the + // head. + skipBody(deliveredPrefix); + currentBody = + new BoundedSegmentStream(header.bufferLength - deliveredPrefix, deliveredPrefix); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + /** Review Comment: Compressed it to one line; I'd keep the "no skipping, the stream already sits on the next header" part since that invariant isn't obvious from the code. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java: ########## @@ -22,42 +22,124 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.apache.flink.util.Preconditions.checkNotNull; /** - * Sealed container for fetched recovered channel-state data. + * Sealed container for recovered channel-state data written to spill files. + * + * <p>Holds a list of file paths (in write order). Segment boundaries are self-described in disk + * segment headers ([4B gateIdx][4B channelIdx][4B bufferLength]), so no in-memory segment locator + * table is maintained. The reader scans files sequentially, reading each 12-byte header to obtain + * the channel info and body length. + * + * <p>The file list grows as the writer rotates to new files (one rotation per 64 MB soft limit), Review Comment: I think it is caused by multi-refactor. Good catch — reworded: the list is fixed at construction (the growth happens earlier, in the writer), and dropped the stale "Mutations" paragraph as well. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReader.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.annotation.Internal; + +import java.io.Closeable; +import java.io.InputStream; +import java.util.Collections; +import java.util.Optional; + +/** + * Forward reader over a {@link FetchedChannelState}'s spill files. This is our own segment reader, + * on purpose <em>not</em> a Java {@link java.util.Iterator}: our access pattern ("a body must be + * fully read before the next segment", "body ownership is handed to the consumer", "consume and + * commit are separate steps") does not fit the {@code hasNext/next} contract. + * + * <p>This interface is the contract callers depend on; {@link FetchedChannelStateReaderImpl} holds + * the implementation (the live file stream, the two progress positions, the bounded body view). + * + * <p>Reading is strictly sequential: a reader is positioned once (offset 0 for the root reader, or + * the committed position for a {@link #snapshot()}), then consumes forward only via {@link + * #nextSegment()}. It never seeks backward and never re-positions mid-iteration. + * + * <p>The drain thread reads the root reader front to back and commits via {@link + * SpillSegment#commit()}; each checkpoint derives a fresh {@link #snapshot()} that resumes from the + * committed position. {@link #snapshot()} and {@link SpillSegment#commit()} must be called under + * the drainer lock; disk reads happen outside it. + */ +@Internal +public interface FetchedChannelStateReader extends Closeable { + + /** + * Advances to the next segment and returns it, or {@link Optional#empty()} when no segment + * remains. Advancing and probing are one step; there is no separate {@code hasNext}. + * + * <p>Entry rule (the first call is exempt): the previous segment's body must be fully read, + * otherwise this is a contract violation and fails loud (no skip-ahead). + */ + Optional<SpillSegment> nextSegment(); + + /** + * Derives an independent resume point starting from the committed position. The snapshot holds Review Comment: The committed position is the delivered boundary recorded by `SpillSegment#commit()` — before anything is committed it's simply the reader's start position; now defined in the interface javadoc. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; + private final List<Path> files; + + /** Live read position; {@code readOffset} is where the open stream physically sits. */ + private final Position current; + + /** Delivered boundary; {@link SpillSegment#commit()} advances it from {@link #current}. */ + private final Position committed; + + /** Open stream over {@code current.fileIndex}, or {@code null} before the first read. */ + @Nullable private InputStream fileStream; Review Comment: Renamed. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/RecoveredChannelStateHandler.java: ########## @@ -214,9 +234,368 @@ public void recover( } } +/** + * Intermediate abstract base for the two spilling variants. Owns the on-disk spill format end to + * end: a single reusable {@link DataOutputSerializer} accumulates one channel's segment, the + * segment header is backfilled with the body length at seal time, and sealed segments are flushed + * to the current file stream with 64 MB-bounded rotation. + * + * <h3>Disk format</h3> + * + * <pre> + * [ 4B BE int: gate idx ] segment header: written once per channel segment + * [ 4B BE int: channel idx ] + * [ 4B BE int: buffer length ] segment body byte count (backfilled at segment seal) + * [ 4B BE int: record length ] repeated for every record in this segment + * [ N bytes: serialized record ] + * [ 4B BE int: gate idx ] next segment header (channel switch or post-rotation) + * ... + * </pre> + * + * <p>The body byte count is only known after the whole segment is written, so each segment is first + * accumulated in {@link #segmentSerializer} (header written at open with a zero placeholder) and + * {@link DataOutputSerializer#writeIntUnsafe} backfills the length at seal. A segment is one + * uninterrupted run of records for a single channel; file rotation happens only after a segment is + * fully sealed, so a segment never crosses a file boundary. + */ +abstract class AbstractSpillingHandler extends AbstractInputChannelRecoveredStateHandler { + + /** Byte offset of the {@code bufferLength} field within a segment's header. */ + static final int BUFFER_LENGTH_HEADER_OFFSET = 2 * Integer.BYTES; + + /** Total size of the segment header in bytes: gateIdx + channelIdx + bufferLength. */ + static final int SEGMENT_HEADER_BYTES = 3 * Integer.BYTES; + + final String[] spillTmpDirectories; + + public static final long DEFAULT_SPILL_FILE_SIZE_BYTES = 64L * 1024 * 1024; + + /** Soft per-file size bound that triggers rotation between segments. */ + private final long maxFileSizeBytes; + + /** + * Accumulates the current segment: the header followed by the body, which is either + * length-prefixed filtered records or verbatim pass-through bytes, depending on the subclass. + * Reused across segments via {@code clear()}. + */ + private final DataOutputSerializer segmentSerializer = new DataOutputSerializer(256); + + /** + * Spill files written so far, in order. The {@link FetchedChannelState} handoff is built from + * this list once writing is sealed; an empty list means the handler never spilled any bytes, so + * it produces no state. + */ + private final List<Path> files = new ArrayList<>(); + + /** + * Unique directory for this handler's spill files; created lazily when the first file opens. + */ + private final Path baseDir; + + /** + * Output stream to the current spill file; tracks the bytes written so far via {@link + * OffsetAwareOutputStream#getLength()} to decide when to rotate. Null before the first segment + * is flushed. + */ + @Nullable private OffsetAwareOutputStream currentStream; + + /** Channel whose segment is currently open; null when no segment is in progress. */ + @Nullable private InputChannelInfo currentChannel; + + @Nullable private FetchedChannelState producedChannelState; + + AbstractSpillingHandler( + InputGate[] inputGates, + InflightDataRescalingDescriptor channelMapping, + String[] spillTmpDirectories, + long maxFileSizeBytes) { + // 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); + checkArgument( + checkNotNull(spillTmpDirectories).length > 0, + "spillTmpDirectories must not be empty"); + checkArgument( + maxFileSizeBytes > 0, "maxFileSizeBytes must be positive: %s", maxFileSizeBytes); + this.spillTmpDirectories = spillTmpDirectories; + this.maxFileSizeBytes = maxFileSizeBytes; + this.baseDir = + Paths.get(spillTmpDirectories[0], "flink-channel-spill-" + UUID.randomUUID()); + } + + /** + * Opens (or switches to) the segment for {@code channelInfo} and returns its buffer for the + * caller to append the body into. The caller must not seal the segment. + */ + DataOutputSerializer segmentSerializerFor(InputChannelInfo channelInfo) throws IOException { + switchChannelIfNeeded(channelInfo); + return segmentSerializer; + } + + private void switchChannelIfNeeded(InputChannelInfo channelInfo) throws IOException { + if (channelInfo.equals(currentChannel)) { + return; + } + if (currentChannel != null) { + sealCurrentSegment(); + } + segmentSerializer.clear(); + segmentSerializer.writeInt(channelInfo.getGateIdx()); + segmentSerializer.writeInt(channelInfo.getInputChannelIdx()); + segmentSerializer.writeInt(0); // bufferLength placeholder + currentChannel = channelInfo; + } + + /** + * Backfills the body length into the segment header and flushes the whole segment to the file + * stream. Empty segments (filtered out entirely, or a zero-byte pass-through) are dropped + * without opening a file, so no empty file is created. + */ + private void sealCurrentSegment() throws IOException { + if (currentChannel == null) { + return; + } + currentChannel = null; + int totalBytes = segmentSerializer.length(); + int bodyBytes = totalBytes - SEGMENT_HEADER_BYTES; + if (bodyBytes == 0) { + return; + } + // Math.toIntExact guards against the unlikely case of a single segment > 2 GB. + segmentSerializer.writeIntUnsafe(Math.toIntExact(bodyBytes), BUFFER_LENGTH_HEADER_OFFSET); + ensureFileOpen(); + currentStream.write(segmentSerializer.getSharedBuffer(), 0, totalBytes); + } + + /** + * Ensures an output stream is ready for the next segment, rotating to a fresh file first if the + * current one reached the size bound. Rotation happens here, between sealed segments, so a + * segment is never split across files. + */ + private void ensureFileOpen() throws IOException { + if (currentStream != null && currentStream.getLength() >= maxFileSizeBytes) { + currentStream.flush(); + currentStream.close(); + currentStream = null; + } + if (currentStream != null) { + return; + } + // create the spill dir on the first file; no-op afterwards + Files.createDirectories(baseDir); + Path filePath = baseDir.resolve("spill-segment-" + files.size() + ".bin"); + currentStream = + new OffsetAwareOutputStream( + new BufferedOutputStream(new FileOutputStream(filePath.toFile())), 0L); + files.add(filePath); + } + + @Override + @Nullable + FetchedChannelState getProducedChannelState() { Review Comment: Yes, intentional — added a TODO referencing FLINK-38544; the wiring lands with the spilling-backend follow-up (#28662). ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; + private final List<Path> files; + + /** Live read position; {@code readOffset} is where the open stream physically sits. */ + private final Position current; + + /** Delivered boundary; {@link SpillSegment#commit()} advances it from {@link #current}. */ + private final Position committed; + + /** Open stream over {@code current.fileIndex}, or {@code null} before the first read. */ + @Nullable private InputStream fileStream; + + /** Size of the file currently open. */ + private long currentFileSize; + + /** Body view of the segment returned by the last {@link #nextSegment()}, or {@code null}. */ + @Nullable private BoundedSegmentStream currentBody; + + private boolean positioned; + private boolean closed; + + FetchedChannelStateReaderImpl(FetchedChannelStateSnapshot snapshot) { + this.snapshot = snapshot; + this.channelState = snapshot.channelState(); + this.files = channelState.files(); + // Must copy the position so that this reader's commits do not mutate the snapshot's state. + this.committed = snapshot.position().copy(); + this.current = committed.copy(); + } + + @Override + public Optional<SpillSegment> nextSegment() { + checkState(!closed, "FetchedChannelStateReader is closed"); + checkState( + currentBody == null || currentBody.remaining() == 0, + "Previous segment body not fully consumed before advancing: %s bytes left", + currentBody == null ? 0 : currentBody.remaining()); + try { + if (!positioned) { + positioned = true; + return firstSegment(); + } + return followingSegment(); + } catch (IOException e) { + throw new RuntimeException("Failed to read segment", e); + } + } + + /** + * First positioning — the only path that may skip bytes. Opens the file at the committed header + * offset and reads the header. A snapshot may resume in the middle of a segment: the committed + * {@code readOffset} says how many body bytes were already delivered, and that prefix is + * skipped so the returned body starts at the not-yet-delivered remainder. If the segment was + * already fully delivered (prefix == whole body), it is exhausted here and we move on to the + * next one. + */ + private Optional<SpillSegment> firstSegment() throws IOException { + // The committed read offset may sit mid-body (after a partial commit), but the header lives + // at segmentStartOffset. Capture how much was already delivered, then rewind the live read + // offset to the header so we open the file there and read the header, not mid-body. + int deliveredPrefix = (int) current.deliveredBodyBytes(); + current.rewindToSegmentStart(); + + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + checkState( + deliveredPrefix <= header.bufferLength, + "Delivered offset %s exceeds segment length %s", + deliveredPrefix, + header.bufferLength); + + if (deliveredPrefix == header.bufferLength) { + // This segment was already fully delivered before the snapshot; nothing remains in it. + // Skip its whole body to reach the next segment's header, then take the steady path. + skipBody(header.bufferLength); + return followingSegment(); + } + + // Discard the already-delivered prefix (the one and only skip in this class), then hand out + // the remainder. alreadyDelivered is carried so commit() records the boundary from the + // head. + skipBody(deliveredPrefix); + currentBody = + new BoundedSegmentStream(header.bufferLength - deliveredPrefix, deliveredPrefix); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + /** + * Steady-state path — no skipping. The previous body was read to its end, so the stream sits + * exactly on this segment's header (or at the current file's end, in which case we roll to the + * next file). Reads the header and returns the whole-body view. + */ + private Optional<SpillSegment> followingSegment() throws IOException { + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + currentBody = new BoundedSegmentStream(header.bufferLength); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + @Override + public FetchedChannelStateSnapshot snapshot() { + checkState(!closed, "FetchedChannelStateReader is closed"); + return new FetchedChannelStateSnapshot(channelState, committed.copy()); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + closed = true; + try { + closeFileStream(); + } finally { + snapshot.release(); + } + } + + // ------------------------------------------------------------------------------------------- + // Sequential IO over the spill files; all of it advances current.readOffset / current.fileIndex + // ------------------------------------------------------------------------------------------- + + /** + * Ensures a file is open with the stream positioned at {@code current}'s read offset, ready to + * read this segment's header. Rolls to the next file when the current one is exhausted. Returns + * false when no segment remains. + * + * <p>{@code current.segmentStartOffset} is set to where the header begins, so a later {@link + * SpillSegment#commit()} records the right segment for the snapshot to resume from. + */ + private boolean openCurrentFile() throws IOException { + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + // Current file fully read: move to the next file's first segment. + closeFileStream(); + current.rollToNextFile(); + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + return false; + } + + /** Reads the 12-byte header at the current read offset; advances past it. */ + private SegmentHeader readHeaderAtCurrent() throws IOException { + byte[] headerBytes = new byte[SEGMENT_HEADER_BYTES]; + readFully(headerBytes); + DataInputStream h = new DataInputStream(new ByteArrayInputStream(headerBytes)); + int gateIdx = h.readInt(); + int channelIdx = h.readInt(); + int bufferLength = h.readInt(); + checkState(bufferLength >= 0, "negative segment length: %s", bufferLength); + return new SegmentHeader(new InputChannelInfo(gateIdx, channelIdx), bufferLength); + } + + /** + * Ensures the file at {@code current.fileIndex} is open with the stream positioned at {@code + * current.readOffset}. If a stream is already open it is left as-is: sequential reading + * guarantees it is already there. + */ + private void openFileAndSeek() throws IOException { + if (fileStream != null) { + return; + } + Path path = files.get(current.fileIndex); + currentFileSize = Files.size(path); + InputStream in = Files.newInputStream(path); + try { + skipOnStream(in, current.readOffset, path); + } catch (IOException e) { + in.close(); + throw e; + } + fileStream = in; + } + + /** Skips {@code count} body bytes on the open stream, advancing the read offset. */ + private void skipBody(long count) throws IOException { + if (count > 0) { + skipOnStream(fileStream, count, files.get(current.fileIndex)); + current.advanceReadOffset(count); Review Comment: Done — `skipOnStream` now looks up the path itself. ########## flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateReaderImpl.java: ########## @@ -0,0 +1,533 @@ +/* + * 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.annotation.Internal; +import org.apache.flink.runtime.checkpoint.channel.FetchedChannelStateReader.SpillSegment; + +import javax.annotation.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Optional; + +import static org.apache.flink.runtime.checkpoint.channel.AbstractSpillingHandler.SEGMENT_HEADER_BYTES; +import static org.apache.flink.util.Preconditions.checkState; + +/** + * The single {@link FetchedChannelStateReader} implementation over a {@link FetchedChannelState}'s + * spill files. + * + * <p>Reading is strictly sequential and never seeks mid-iteration. There is exactly one place that + * skips bytes: the very first {@link #nextSegment()} call, where a snapshot reader started mid-body + * discards the already-delivered prefix to land on the not-yet-delivered remainder. Every later + * call does no skipping at all — the previous body was read to its end, so the stream already sits + * on the next segment's header. This "skip only on first positioning" rule is what keeps the + * steady-state path free of any seek/skip. + * + * <p>The reader holds <b>two</b> {@link Position}s and nothing else duplicates them: + * + * <ul> + * <li>{@code current} — the live read position; its {@code readOffset} is exactly where the open + * file stream sits, advancing as the header and the consumer's body reads consume bytes (the + * latter outside the drainer lock). + * <li>{@code committed} — the delivered boundary; {@link SpillSegment#commit()} advances it from + * {@code current} (under the drainer lock). {@link #snapshot()} derives a new reader from it. + * </ul> + * + * <p>The "previous body fully read before advancing" rule is checked at the {@link #nextSegment()} + * entry (the first call is exempt — there is no previous segment). Body ownership is handed to the + * consumer, so the reader does not track body progress except through {@code current}. + */ +@Internal +final class FetchedChannelStateReaderImpl implements FetchedChannelStateReader { + + private final FetchedChannelStateSnapshot snapshot; + private final FetchedChannelState channelState; + private final List<Path> files; + + /** Live read position; {@code readOffset} is where the open stream physically sits. */ + private final Position current; + + /** Delivered boundary; {@link SpillSegment#commit()} advances it from {@link #current}. */ + private final Position committed; + + /** Open stream over {@code current.fileIndex}, or {@code null} before the first read. */ + @Nullable private InputStream fileStream; + + /** Size of the file currently open. */ + private long currentFileSize; + + /** Body view of the segment returned by the last {@link #nextSegment()}, or {@code null}. */ + @Nullable private BoundedSegmentStream currentBody; + + private boolean positioned; + private boolean closed; + + FetchedChannelStateReaderImpl(FetchedChannelStateSnapshot snapshot) { + this.snapshot = snapshot; + this.channelState = snapshot.channelState(); + this.files = channelState.files(); + // Must copy the position so that this reader's commits do not mutate the snapshot's state. + this.committed = snapshot.position().copy(); + this.current = committed.copy(); + } + + @Override + public Optional<SpillSegment> nextSegment() { + checkState(!closed, "FetchedChannelStateReader is closed"); + checkState( + currentBody == null || currentBody.remaining() == 0, + "Previous segment body not fully consumed before advancing: %s bytes left", + currentBody == null ? 0 : currentBody.remaining()); + try { + if (!positioned) { + positioned = true; + return firstSegment(); + } + return followingSegment(); + } catch (IOException e) { + throw new RuntimeException("Failed to read segment", e); + } + } + + /** + * First positioning — the only path that may skip bytes. Opens the file at the committed header + * offset and reads the header. A snapshot may resume in the middle of a segment: the committed + * {@code readOffset} says how many body bytes were already delivered, and that prefix is + * skipped so the returned body starts at the not-yet-delivered remainder. If the segment was + * already fully delivered (prefix == whole body), it is exhausted here and we move on to the + * next one. + */ + private Optional<SpillSegment> firstSegment() throws IOException { + // The committed read offset may sit mid-body (after a partial commit), but the header lives + // at segmentStartOffset. Capture how much was already delivered, then rewind the live read + // offset to the header so we open the file there and read the header, not mid-body. + int deliveredPrefix = (int) current.deliveredBodyBytes(); + current.rewindToSegmentStart(); + + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + checkState( + deliveredPrefix <= header.bufferLength, + "Delivered offset %s exceeds segment length %s", + deliveredPrefix, + header.bufferLength); + + if (deliveredPrefix == header.bufferLength) { + // This segment was already fully delivered before the snapshot; nothing remains in it. + // Skip its whole body to reach the next segment's header, then take the steady path. + skipBody(header.bufferLength); + return followingSegment(); + } + + // Discard the already-delivered prefix (the one and only skip in this class), then hand out + // the remainder. alreadyDelivered is carried so commit() records the boundary from the + // head. + skipBody(deliveredPrefix); + currentBody = + new BoundedSegmentStream(header.bufferLength - deliveredPrefix, deliveredPrefix); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + /** + * Steady-state path — no skipping. The previous body was read to its end, so the stream sits + * exactly on this segment's header (or at the current file's end, in which case we roll to the + * next file). Reads the header and returns the whole-body view. + */ + private Optional<SpillSegment> followingSegment() throws IOException { + if (!openCurrentFile()) { + return Optional.empty(); + } + SegmentHeader header = readHeaderAtCurrent(); + currentBody = new BoundedSegmentStream(header.bufferLength); + return Optional.of(new Segment(header.channelInfo, currentBody)); + } + + @Override + public FetchedChannelStateSnapshot snapshot() { + checkState(!closed, "FetchedChannelStateReader is closed"); + return new FetchedChannelStateSnapshot(channelState, committed.copy()); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + closed = true; + try { + closeFileStream(); + } finally { + snapshot.release(); + } + } + + // ------------------------------------------------------------------------------------------- + // Sequential IO over the spill files; all of it advances current.readOffset / current.fileIndex + // ------------------------------------------------------------------------------------------- + + /** + * Ensures a file is open with the stream positioned at {@code current}'s read offset, ready to + * read this segment's header. Rolls to the next file when the current one is exhausted. Returns + * false when no segment remains. + * + * <p>{@code current.segmentStartOffset} is set to where the header begins, so a later {@link + * SpillSegment#commit()} records the right segment for the snapshot to resume from. + */ + private boolean openCurrentFile() throws IOException { + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + // Current file fully read: move to the next file's first segment. + closeFileStream(); + current.rollToNextFile(); + if (current.fileIndex >= files.size()) { + return false; + } + openFileAndSeek(); + if (current.readOffset < currentFileSize) { + current.startSegmentHere(); + return true; + } + return false; + } + + /** Reads the 12-byte header at the current read offset; advances past it. */ + private SegmentHeader readHeaderAtCurrent() throws IOException { + byte[] headerBytes = new byte[SEGMENT_HEADER_BYTES]; + readFully(headerBytes); + DataInputStream h = new DataInputStream(new ByteArrayInputStream(headerBytes)); + int gateIdx = h.readInt(); + int channelIdx = h.readInt(); + int bufferLength = h.readInt(); + checkState(bufferLength >= 0, "negative segment length: %s", bufferLength); + return new SegmentHeader(new InputChannelInfo(gateIdx, channelIdx), bufferLength); + } + + /** + * Ensures the file at {@code current.fileIndex} is open with the stream positioned at {@code + * current.readOffset}. If a stream is already open it is left as-is: sequential reading + * guarantees it is already there. + */ + private void openFileAndSeek() throws IOException { + if (fileStream != null) { + return; + } + Path path = files.get(current.fileIndex); + currentFileSize = Files.size(path); + InputStream in = Files.newInputStream(path); + try { + skipOnStream(in, current.readOffset, path); + } catch (IOException e) { + in.close(); + throw e; + } + fileStream = in; + } + + /** Skips {@code count} body bytes on the open stream, advancing the read offset. */ + private void skipBody(long count) throws IOException { + if (count > 0) { + skipOnStream(fileStream, count, files.get(current.fileIndex)); + current.advanceReadOffset(count); + } + } + + /** Skips exactly {@code count} bytes on {@code in}, failing loud if the file ends early. */ + private void skipOnStream(InputStream in, long count, Path path) throws IOException { Review Comment: Is this comment in the wrong line? IIUC, it should be commented on `private void readFully(byte[] buf)`. Done — switched to `IOUtils.readFully`, wrapping the exception to keep the file/offset context in the message. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
