rkhachatryan commented on PR #29065:
URL: https://github.com/apache/flink/pull/29065#issuecomment-5588018359
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:
```java
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
`unregisterCloseable`d 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.
--
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]
