[FLINK-40520][checkpointing] Release spilled channel state on recovery abort between fetch and drain

[1996fanrui]

What is the purpose of the change

Spilled channel-state files produced during checkpointing during recovery were released only when drain()
ran. Several fetch→drain abort paths never reach drain (readInputData throws after a file was spilled; the
requestPartitions or trigger-install mail is rejected or dropped; thenRunAsync(drain) is rejected after the
channelIOExecutor shut down), so the produced FetchedChannelState was never closed and its spill files
survived until TaskManager shutdown, accumulating across recovery-failure loops on a long-lived pooled TM.

The fix deletes the spill files at the two places that can still own them on an abort: readInputData when
the fetch fails before handing the state off (it deletes the handler's whole spill directory, so this holds
even when stateHandler.close() itself failed and no FetchedChannelState was built), and the task's
resourceCloser (cleanUp()) for everything after the hand-off. Hooking into task cleanup rather than the recovery future chain matters because a mail
dropped on mailbox close never completes its future, so no future callback would ever fire for it.

Brief change log

• [FLINK-40520] Delete the spilled files on abort:
  • readInputData deletes the spilling 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). FetchedChannelState.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.

Verifying this change

This change added a test:

• SequentialChannelStateReaderImplTest asserts the spill files and their directory are deleted when readInputData aborts after spilling at least one file (fails without the fix).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: yes (channel-state recovery)
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[flinkbot]

CI report:

• 64ffa4d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[rkhachatryan]

We might as well pass CloseableRegistry cancelables here instead (and skip registering state in the caller), right?

Not sure if interface change worth it though.

[rkhachatryan]

Actually, is this (closeQuietly) here enough? What if an exception happens before stateHandler assigns a value to producedChannelState?

I think a proper fix would be to either:

1. Expose some cleanupOnFailure method on stateHandler that'd clean up any temporary files
2. Pass CloseableRegistry to stateHandler and use it there for all the resources (files/state)

(1) seems more explicit to me.

WDYT?

[1996fanrui]

Done — the spilling handler now takes the CloseableRegistry and registers its spill files itself, covering both.

[rkhachatryan]

Can we have race conditions here? 🤔

I.e. a task in recovery is being cancelled; files is not updated yet by the recovery; or the updates are not visible to the closing thread?

Maybe just delete the directory instead (if it's exclusive)? Or can this cause some failure of the recovering thread?

[1996fanrui]

Good catch — switched to deleteDirectoryQuietly(best-effort/quiet)

[rkhachatryan]

Review generated with Claude Code (posted by @rkhachatryan; findings reviewed but the analysis below is Claude's, so please double-check the reasoning).

Four observations, all stemming from the cleanup hook being a one-shot whole-directory delete rather than the refcount-aware close the PR description implies.

1. The hook bypasses FetchedChannelState's refcount contract (medium)

FetchedChannelState documents that "files are deleted only when the last lifecycle grant is released", but the hook deletes baseDir unconditionally. cancelables is registered on resourceCloser after channelIOExecutor::shutdown and this::shutdownAsyncThreads (StreamTask:457/491/492), and AutoCloseableRegistry closes in reverse order — so cancelables.close() runs before the channel-IO executor is drained. cancel() also calls cancelables.close() directly (StreamTask:1395), by design, to interrupt blocking I/O.

Scenario: an operator fails shortly after recovery while drain() is still running on channelIOExecutor. The directory is deleted; FetchedChannelStateReaderImpl.openFileAndSeek() opens files lazily one at a time, so the next Files.newByteChannel throws NoSuchFileException, drain() reports it via asyncExceptionHandler.handleAsyncException → failExternally, adding a spurious failure that can mask the real root cause. Same for an async recovery-checkpoint snapshot reader still writing spill segments to checkpoint storage.

Reordering the registration doesn't help (cancel closes the registry explicitly regardless). Suggestion: hand ownership to FetchedChannelState::close so the closed flag is at least set, and gate the reporting in drain() (StreamTask:1102) and the exceptionally handler on the existing canceled / !isRunning state — log at debug when the task is already going down — or throw a typed abort exception that drain() treats as expected.

2. One-shot registration lets post-abort spill files leak permanently (medium)

Registration is guarded by spillCleanupRegistered, but file creation is not: every ensureFileOpen() re-runs Files.createDirectories(baseDir).

Scenario: the task is cancelled while readInputData is still fetching a large channel state (>64 MB, i.e. after at least one rotation). cancel() → cancelables.close() fires and removes the hook, deleting baseDir. The fetch loop on channelIOExecutor is not interrupted and never touches the registry again (the flag is already true, so no further registerCloseable — hence no IOException to abort it either), so it recreates baseDir and keeps writing spill-segment-N.bin. Those files are never deleted — the exact leak this PR fixes, now permanent until TM shutdown.

Suggested fix — make the hook stateful instead of one-shot:

private final class SpillCleanup implements Closeable {
    private volatile boolean aborted;
    @Override public void close() {
        aborted = true;                                  // set before deleting
        FileUtils.deleteDirectoryQuietly(baseDir.toFile());
    }
}

and in ensureFileOpen(), after files.add(filePath), bail out if aborted: close the stream, delete what was just written, throw IOException. That terminates the fetch loop and closes the write-after-delete window (files created before the hook fires are removed by it; files created after see the flag). Re-registering per file would also work — registerCloseable on a closed registry throws IOException and closes the argument (AbstractAutoCloseableRegistry:89) — but it accumulates entries.

3. The finally-close described in the PR text isn't in the diff (low)

The description says the produced FetchedChannelState is force-closed in a finally in readInputData, but that code isn't there. On abort, closeInternal() has already built the state and called acquire(), and nothing releases it, so cleanup is deferred to whole-task teardown rather than happening at the abort point. Conversely, on the happy path the hook is never unregisterCloseabled after ownership transfers, so cancelables keeps a stale entry (and a strong reference to baseDir) for the task's whole lifetime and re-deletes the directory at teardown.

Suggestion: in closeInternal(), after producedChannelState = new FetchedChannelState(files); acquire();, unregister the hook and register producedChannelState itself (it is Closeable and refcount-aware), then add the described finally-close.

4. Javadoc overstates what is registered (low)

SequentialChannelStateReader:38 — the @param cancelables javadoc says the registry is what "the spilling handler registers its spill files with"; it actually receives a single whole-directory delete hook, and only in the spilling modes. The parameter is also mandatory/non-null on the NoSpillingHandler path where it is entirely unused. Same wording appears in the two new inline comments in SequentialChannelStateReaderImpl and StreamTask.fetchChannelState.

Checked and looks fine

• No other implementors/callers of readInputData beyond NO_OP, SequentialChannelStateReaderImpl, StreamTask, and the two updated tests.
• ChannelStateFilteringHandler.createFromContext returns null for zero gates, so the new test's comment about hitting SpillingNoFilteringHandler is accurate; RecordFilterContext arg order matches the constructor and bufferSize (10/20) satisfies memorySegmentSize > 0.
• assumeTrue(stateParLevel > 0 && parLevel > 0) still leaves 3 of the 5 parameter combinations running, so the new test isn't silently skipped.
• Registration happens before Files.createDirectories(baseDir) and only when currentStream == null, so no fd leak if registerCloseable throws on an already-closed registry.

[1996fanrui]

Spilled channel-state cleanup: ownership model

Fetch and drain run on channelIOExecutor, which is never interrupted on cancel (only a graceful
shutdown(); cancel interrupts just the main task thread); the steps in between (requestPartitions, trigger
install) are mailbox mails. Each stage owns its spill state: it either hands ownership to the next stage or
fails and cleans up. Cancellation does not always surface as a stage failure, though: a mail already queued
when the mailbox closes is dropped and its future never completes, so no future callback can react. Hence the
fetched state is also registered with the task's resourceCloser, and cleanUp() is the backstop for
everything after the fetch.

Stage	Own run fails (throws)	Cancelled mid-run	Runs successfully
1 — fetch (readInputData)	catch deletes the handler's spill directory (also when close() failed before building the FetchedChannelState)	Teardown makes it throw (→ own failure, self-clean) or it finishes first (→ handed to stage 2)	Registers the state with resourceCloser; returns it to stage 2
2 — gap (requestPartitions + buildDrainer, mailbox)	Task fails → cleanUp() closes the state	execute rejected (mailbox closed) → same as failure; mail dropped → future never completes → cleanUp() closes the state	Builds the drainer (takes a grant); trigger-install mail hands to stage 3, same reject/drop coverage
3 — drain	try-with-resources close() → release deletes the files	Not interrupted → completes or throws; both go through close()	close() → release; files deleted normally

Note: FetchedChannelState.close() is idempotent, so after a normal drain cleanUp() is a no-op; a fetch
finishing after cleanUp() is closed by the registry on registration. thenRunAsync(drain) rejected after
executor shutdown is covered the same way: the executor is only shut down after recovery completed or in
cleanUp().

[rkhachatryan]

I think we are back to this issue now:
the failure might happen before the AbstractSpillingHandler.producedChannelState is assigned.

[1996fanrui]

Right, this brought the original issue back.

Fixed with your option (1) from that thread: AbstractSpillingHandler#discardSpilledFiles() deletes the whole spill directory, and the catch in readInputData calls it, so it no longer depends on producedChannelState having been assigned.

I originally wanted to avoid adding a method on the handler for this, but after the detour it turned out to be the simplest fix anyway.

[rkhachatryan]

LGTM

[1996fanrui]

Thanks for the review and suggestion, merging