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
The following commit(s) were added to refs/heads/master by this push:
new dd33da5a2da [FLINK-40520][checkpointing] Release spilled channel state
on recovery abort between fetch and drain
dd33da5a2da is described below
commit dd33da5a2da017f1020258c9d73ca84b968c64d5
Author: Rui Fan <[email protected]>
AuthorDate: Wed Sep 9 23:48:34 2026 +0200
[FLINK-40520][checkpointing] Release spilled channel state on recovery
abort between fetch and drain
Spill files produced by readInputData were only deleted once drain() ran,
so any
abort between fetch and drain leaked them until TaskManager shutdown.
- readInputData deletes the handler's spill directory if it fails before
handing the
state off. This does not depend on the produced FetchedChannelState
having been
built, since stateHandler.close() itself may be what failed.
- fetchChannelState registers the fetched state with the task's
resourceCloser, so
cleanUp() deletes the spill files whenever recovery aborts afterwards
(drainer
never built, mailbox mail rejected or dropped, drain() never scheduled).
close()
is idempotent, so a completed drain makes this a no-op; a fetch that
finishes
after cleanUp() is closed by the registry on the spot.
---
.../channel/RecoveredChannelStateHandler.java | 14 ++++
.../channel/SequentialChannelStateReaderImpl.java | 8 ++
.../flink/streaming/runtime/tasks/StreamTask.java | 6 ++
.../SequentialChannelStateReaderImplTest.java | 94 ++++++++++++++++++++++
4 files changed, 122 insertions(+)
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 03ac7f8309c..427f1d5cc9e 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
@@ -36,6 +36,8 @@ import
org.apache.flink.runtime.io.network.partition.CheckpointedResultPartition
import org.apache.flink.runtime.io.network.partition.consumer.InputChannel;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import
org.apache.flink.runtime.io.network.partition.consumer.RecoveredInputChannel;
+import org.apache.flink.util.FileUtils;
+import org.apache.flink.util.IOUtils;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -413,6 +415,18 @@ abstract class AbstractSpillingHandler extends
AbstractInputChannelRecoveredStat
return files;
}
+ /**
+ * Deletes the spill directory with everything written so far, whether or
not {@link
+ * #closeInternal()} has built {@link #producedChannelState}. Best-effort,
never throws.
+ */
+ void discardSpilledFiles() {
+ IOUtils.closeQuietly(currentStream);
+ currentStream = null;
+ // Marks the state closed for anyone still holding it before the
directory goes away.
+ IOUtils.closeQuietly(producedChannelState);
+ FileUtils.deleteDirectoryQuietly(baseDir.toFile());
+ }
+
/**
* Seals the open segment and the file stream, then builds the {@link
FetchedChannelState}
* handoff from the written files. Produces nothing if no bytes were ever
spilled.
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 9284c226a59..9b35f1209c1 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
@@ -103,6 +103,14 @@ public class SequentialChannelStateReaderImpl implements
SequentialChannelStateR
// 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());
+ } catch (Throwable t) {
+ // The state was not handed off, so no drainer will release its
spill files: delete
+ // them here. This does not rely on the produced state having been
built, since
+ // stateHandler.close() itself may be what failed.
+ if (stateHandler instanceof AbstractSpillingHandler) {
+ ((AbstractSpillingHandler) stateHandler).discardSpilledFiles();
+ }
+ throw t;
}
}
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index 113f3ea046d..7bccacd79dd 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -1081,6 +1081,12 @@ public abstract class StreamTask<OUT, OP extends
StreamOperator<OUT>>
Optional<FetchedChannelState> state =
reader.readInputData(inputGates,
createRecordFilterContext());
if (state.isPresent()) {
+ // The task owns the spill files until a drainer releases
them; registering here
+ // deletes them on cleanUp() if recovery aborts anywhere
between fetch and drain
+ // (the drainer never gets built, a mailbox mail is rejected
or dropped, drain() is
+ // never scheduled). close() is idempotent, so a completed
drain makes this a no-op.
+ // If cleanUp() already ran, the registry closes the state
right here instead.
+ resourceCloser.registerCloseable(state.get());
LOG.info(
"Fetched and filtered the recovered channel state into
{} spill file(s).",
state.get().files().size());
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
index dc09e35cffc..a7c3b8b26e4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/channel/SequentialChannelStateReaderImplTest.java
@@ -18,8 +18,11 @@
package org.apache.flink.runtime.checkpoint.channel;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.checkpoint.InflightDataRescalingDescriptor;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
+import org.apache.flink.runtime.checkpoint.StateObjectCollection;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
@@ -40,6 +43,7 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.InputChannelStateHandle;
import org.apache.flink.runtime.state.ResultSubpartitionStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
+import org.apache.flink.runtime.state.testutils.EmptyStreamStateHandle;
import org.apache.flink.streaming.runtime.io.recovery.RecordFilterContext;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
@@ -51,10 +55,13 @@ import
org.apache.flink.shaded.guava33.com.google.common.io.Closer;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -65,6 +72,7 @@ import java.util.Optional;
import java.util.Random;
import java.util.function.BiFunction;
import java.util.function.Function;
+import java.util.stream.Stream;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.toList;
@@ -73,6 +81,8 @@ import static java.util.stream.IntStream.range;
import static
org.apache.flink.runtime.state.ChannelStateHelper.castToInputStateCollection;
import static
org.apache.flink.runtime.state.ChannelStateHelper.castToOutputStateCollection;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
/** {@link SequentialChannelStateReaderImpl} Test. */
@ExtendWith(ParameterizedTestExtension.class)
@@ -107,6 +117,8 @@ public class SequentialChannelStateReaderImplTest {
@Parameter(value = 5)
public int bufferSize;
+ @TempDir Path tempDir;
+
private ChannelStateSerializer serializer;
private Random random;
private int buffersPerChannel;
@@ -156,6 +168,88 @@ public class SequentialChannelStateReaderImplTest {
});
}
+ @TestTemplate
+ void testReadInputDataDeletesSpillFilesOnAbort() throws Exception {
+ // The empty-state parameter combos spill nothing, so there is nothing
to assert; skip them.
+ assumeTrue(stateParLevel > 0 && parLevel > 0);
+
+ Map<InputChannelInfo, List<byte[]>> inputChannelsData =
+ generateState(InputChannelInfo::new);
+
+ // read#1 spills valid input-channel state; read#2's delegate fails,
so readInputData aborts
+ // after spilling.
+ List<InputChannelStateHandle> validInputState =
+ writePermuted(inputChannelsData, Collections.emptyMap()).f0;
+ InputChannelStateHandle failingUpstreamState =
+ new InputChannelStateHandle(
+ new InputChannelInfo(0, 0),
+ new FailingStreamStateHandle(),
+ Collections.singletonList(0L));
+ TaskStateSnapshot snapshot =
+ new TaskStateSnapshot(
+ Collections.singletonMap(
+ new OperatorID(),
+ OperatorSubtaskState.builder()
+ .setInputChannelState(
+
castToInputStateCollection(validInputState))
+ .setUpstreamOutputBufferState(
+ new StateObjectCollection<>(
+
Collections.singletonList(
+
failingUpstreamState)))
+ .build()));
+
+ SequentialChannelStateReader reader = new
SequentialChannelStateReaderImpl(snapshot);
+ // Empty inputConfigs -> no filtering handler -> the plain
SpillingNoFilteringHandler path.
+ RecordFilterContext spillingContext =
+ new RecordFilterContext(
+ new RecordFilterContext.InputFilterConfig[0],
+ InflightDataRescalingDescriptor.NO_RESCALE,
+ 0,
+ parLevel,
+ new String[] {tempDir.toString()},
+ true,
+ bufferSize);
+
+ withInputGates(
+ gates -> {
+ assertThatThrownBy(() -> reader.readInputData(gates,
spillingContext))
+ .isInstanceOf(IOException.class);
+
+ // The abort deleted the spill files and the per-recovery
directory.
+ assertThat(listSpillFiles(tempDir)).isEmpty();
+ assertThat(listSpillDirs(tempDir)).isEmpty();
+ });
+ }
+
+ /**
+ * A stream state handle whose input stream cannot be opened, used to
abort a channel-state read
+ * mid-recovery. A named (non-anonymous) class so it does not trip {@code
+ * StateHandleSerializationTest}, which forbids anonymous {@code
StateObject} subclasses.
+ */
+ private static final class FailingStreamStateHandle extends
EmptyStreamStateHandle {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public FSDataInputStream openInputStream() throws IOException {
+ throw new IOException("injected channel-state read failure");
+ }
+ }
+
+ private static List<Path> listSpillDirs(Path root) throws IOException {
+ try (Stream<Path> entries = Files.list(root)) {
+ return entries.filter(Files::isDirectory)
+ .filter(p ->
p.getFileName().toString().startsWith("flink-channel-spill-"))
+ .collect(toList());
+ }
+ }
+
+ private static List<Path> listSpillFiles(Path root) throws IOException {
+ try (Stream<Path> entries = Files.walk(root)) {
+ return entries.filter(p ->
p.getFileName().toString().endsWith(".bin"))
+ .collect(toList());
+ }
+ }
+
private Map<ResultSubpartitionInfo, List<Buffer>> collectBuffers(
BufferWritingResultPartition[] resultPartitions) throws
IOException {
Map<ResultSubpartitionInfo, List<Buffer>> actual = new HashMap<>();
