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 71a00195a4bbaee8a9b39b193d4a160201bcdb90 Author: Rui Fan <[email protected]> AuthorDate: Mon Jul 6 01:49:31 2026 +0200 [FLINK-39524][checkpoint] Add FetchedChannelState spill-file container Sealed container over an ordered List<Path> of spill files with an acquire()/release() ref-counted lifecycle that deletes the files when the last grant is released (cleanedUp guard); close() forces cleanup. Deviation from the plan's commit scope: the reader() entry point and FetchedChannelStateSnapshot are deferred to the forward-only-reader commit of this PR — both hard-depend on FetchedChannelStateReader / FetchedChannelStateReaderImpl.Position, which do not exist yet. Everything present here is byte-identical to its final form. FetchedChannelStateRefCountTest is likewise deferred (it needs the TestSpillWriter from the spill-writing-handlers commit). --- .../checkpoint/channel/FetchedChannelState.java | 102 +++++++++-- .../channel/SequentialChannelStateReaderImpl.java | 3 +- .../channel/FetchedChannelStateTest.java | 189 +++++++++++++++++++++ 3 files changed, 278 insertions(+), 16 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java index 6e28c525dbb..1137578a2cf 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java @@ -22,42 +22,114 @@ import org.apache.flink.annotation.VisibleForTesting; 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), + * and is sealed on writer close. + * + * <p>File lifecycle is managed by {@link #acquire()} / {@link #release()} reference counting. Files + * are deleted only when the last lifecycle grant is released (i.e. when both the main reader and + * all snapshot readers have finished). * - * <p>FLINK-38544 transitional in-memory placeholder: the in-memory recovery backend keeps all - * recovered buffers inside the physical channels' own queues (pushed in one shot at conversion - * time), so there is no spill file and nothing to hand out or clean up here. This container carries - * only the "there is state to recover" signal that {@link - * SequentialChannelStateReader#readInputData} returns to the {@code StreamTask} recovery link; the - * lifecycle and file APIs are no-ops until the spilling backend lands and replaces this with a - * real, file-backed container. + * <p>Mutations (file list appends) are single-writer and intentionally unsynchronized; callers must + * serialize them via the channel IO executor. */ @Internal public final class FetchedChannelState implements Closeable { + /** Ordered list of spill file paths, one entry per physical file. Sealed at construction. */ + private final List<Path> files; + + // close() and release() may be called from different threads; volatile ensures visibility. private volatile boolean closed = false; - FetchedChannelState() {} + private final AtomicInteger refCount = new AtomicInteger(0); + + private final AtomicBoolean cleanedUp = new AtomicBoolean(false); + + /** + * Wraps an already-written, ordered list of spill files. The list is sealed; it never grows. + */ + FetchedChannelState(List<Path> files) { + this.files = new ArrayList<>(checkNotNull(files)); + } + + // ------------------------------------------------------------------------------------------- + // Read-phase API (called by the reader after the writer is sealed) + // ------------------------------------------------------------------------------------------- - /** Returns the ordered list of spill file paths; empty for the in-memory backend. */ + /** Returns the ordered list of spill file paths. Read-only view. */ public List<Path> files() { - return Collections.emptyList(); + return Collections.unmodifiableList(files); } - /** Acquires a lifecycle grant; no-op for the in-memory backend (no files to keep alive). */ - public void acquire() {} + // ------------------------------------------------------------------------------------------- + // Lifecycle + // ------------------------------------------------------------------------------------------- - /** Releases a lifecycle grant; no-op for the in-memory backend (no files to delete). */ - public void release() throws IOException {} + /** Acquires a lifecycle grant for a reader or handoff owner. */ + public void acquire() { + refCount.incrementAndGet(); + } + /** + * Releases a lifecycle grant. When the last grant is released (refCount reaches zero), all + * spill files are deleted. This preserves the invariant that files exist for the lifetime of + * all readers (drain + snapshot) and are cleaned up exactly once when the last reader finishes. + */ + public void release() throws IOException { + if (refCount.decrementAndGet() == 0) { + if (cleanedUp.compareAndSet(false, true)) { + closed = true; + deleteAllFiles(); + } + } + } + + /** Forces cleanup even when lifecycle grants are still outstanding. */ @Override public void close() throws IOException { + if (closed) { + return; + } closed = true; + if (cleanedUp.compareAndSet(false, true)) { + deleteAllFiles(); + } + } + + private void deleteAllFiles() throws IOException { + IOException firstError = null; + for (Path file : files) { + try { + Files.deleteIfExists(file); + } catch (IOException e) { + if (firstError == null) { + firstError = e; + } else { + firstError.addSuppressed(e); + } + } + } + if (firstError != null) { + throw firstError; + } } @VisibleForTesting 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 e354d1ddf8b..2bfad5d6133 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,6 +33,7 @@ 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; @@ -110,7 +111,7 @@ public class SequentialChannelStateReaderImpl implements SequentialChannelStateR // only signals "there is state to recover". The spilling backend returns a real, // file-backed container here. return filterContext.isCheckpointingDuringRecoveryEnabled() && readAny - ? Optional.of(new FetchedChannelState()) + ? Optional.of(new FetchedChannelState(Collections.emptyList())) : Optional.empty(); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateTest.java new file mode 100644 index 00000000000..33429543dab --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelStateTest.java @@ -0,0 +1,189 @@ +/* + * 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.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.Collections; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for {@link FetchedChannelState} lifecycle and file list management: reference counting + * (acquire/release pairing, zero-triggered file deletion, release-past-zero no-op), forced {@link + * FetchedChannelState#close()} cleanup, and the unmodifiable ordered file list. + */ +class FetchedChannelStateTest { + + @TempDir Path tempDir; + + @Test + void testInitialStateIsEmpty() { + FetchedChannelState state = new FetchedChannelState(Collections.emptyList()); + assertThat(state.files()).isEmpty(); + assertThat(state.isClosed()).isFalse(); + } + + @Test + void testFileListPreservesOrder() throws IOException { + Path file0 = tempDir.resolve("spill-0.bin"); + Path file1 = tempDir.resolve("spill-1.bin"); + + try (FetchedChannelState state = new FetchedChannelState(Arrays.asList(file0, file1))) { + assertThat(state.files()).containsExactly(file0, file1); + } + } + + @Test + void testFilesListIsUnmodifiable() throws IOException { + try (FetchedChannelState state = + new FetchedChannelState(Collections.singletonList(tempDir.resolve("f0.bin")))) { + assertThatThrownBy(() -> state.files().add(tempDir.resolve("f1.bin"))) + .isInstanceOf(UnsupportedOperationException.class); + } + } + + @Test + void testAcquireReleaseDoesNotDeleteFilesBeforeLastRelease() throws IOException { + Path realFile = tempDir.resolve("spill-0.bin"); + realFile.toFile().createNewFile(); + FetchedChannelState state = new FetchedChannelState(Collections.singletonList(realFile)); + + state.acquire(); + state.acquire(); + + state.release(); + // File must still exist after first release. + assertThat(realFile.toFile()).exists(); + + state.release(); + // Last release should delete the file. + assertThat(realFile.toFile()).doesNotExist(); + assertThat(state.isClosed()).isTrue(); + } + + @Test + void testCloseDeletesAllFiles() throws IOException { + Path file0 = tempDir.resolve("f0.bin"); + Path file1 = tempDir.resolve("f1.bin"); + file0.toFile().createNewFile(); + file1.toFile().createNewFile(); + + FetchedChannelState state = new FetchedChannelState(Arrays.asList(file0, file1)); + + state.close(); + + assertThat(file0.toFile()).doesNotExist(); + assertThat(file1.toFile()).doesNotExist(); + assertThat(state.isClosed()).isTrue(); + } + + @Test + void testCloseIsIdempotent() throws IOException { + FetchedChannelState state = new FetchedChannelState(Collections.emptyList()); + state.close(); + assertThat(state.isClosed()).isTrue(); + // Second close must not throw. + state.close(); + assertThat(state.isClosed()).isTrue(); + } + + @Test + void testCloseAfterReleaseIsIdempotent() throws IOException { + FetchedChannelState state = new FetchedChannelState(Collections.emptyList()); + state.acquire(); + state.release(); + assertThat(state.isClosed()).isTrue(); + // close() after last release must be a no-op (no double-delete attempt). + state.close(); + assertThat(state.isClosed()).isTrue(); + } + + @Test + void testReleaseAfterZeroIsNoOp() throws IOException { + FetchedChannelState state = newStateWithData(); + // Release the single handoff grant the produced state already holds. + state.release(); + assertFilesDeleted(state); + + // Extra releases past zero must be a no-op. + state.release(); + state.release(); + assertFilesDeleted(state); + } + + @Test + void testBalancedAcquireReleaseDeletesOnlyOnLastRelease() throws IOException { + FetchedChannelState state = newStateWithData(); + + // The produced state already holds one handoff grant. + state.acquire(); + state.acquire(); + + state.release(); + state.release(); + assertFilesExist(state); + + state.release(); + assertFilesDeleted(state); + } + + @Test + void testForceCloseCleansFilesAndToleratesLateRelease() throws IOException { + FetchedChannelState state = newStateWithData(); + // The produced state already holds one handoff grant. + state.acquire(); + assertFilesExist(state); + + state.close(); + assertFilesDeleted(state); + + // Double close must be a no-op. + state.close(); + + // Late release after close must not re-delete or throw. + state.release(); + assertFilesDeleted(state); + } + + private FetchedChannelState newStateWithData() throws IOException { + try (TestSpillWriter writer = new TestSpillWriter(tempDir)) { + writer.writeRecord(new InputChannelInfo(0, 0), new byte[] {1, 2, 3}, 3); + writer.writeRecord(new InputChannelInfo(0, 1), new byte[] {4, 5}, 2); + return writer.getChannelState(); + } + } + + private static void assertFilesExist(FetchedChannelState state) { + for (Path file : state.files()) { + assertThat(file.toFile()).exists(); + } + } + + private static void assertFilesDeleted(FetchedChannelState state) { + for (Path file : state.files()) { + assertThat(file.toFile()).doesNotExist(); + } + } +}
