1996fanrui commented on code in PR #28661:
URL: https://github.com/apache/flink/pull/28661#discussion_r3736227164
##########
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;
Review Comment:
Yes — added a 1 MB soft bound: a new segment now starts on a channel switch
or once the open one outgrew it.
--
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]
