Rui Fan created FLINK-40520:
-------------------------------
Summary: Recovered channel-state spill files leak when recovery
aborts between fetch and drain
Key: FLINK-40520
URL: https://issues.apache.org/jira/browse/FLINK-40520
Project: Flink
Issue Type: Sub-task
Components: Runtime / Checkpointing
Reporter: Rui Fan
Assignee: Rui Fan
Spilled channel-state files are cleaned up only when `drain(drainer)` runs [1].
But several fetch→drain abort paths never reach `drain`, so the files are never
released — nobody else owns cleanup (the state/drainer is never registered with
`cancelables`, and `recoveryCheckpointTrigger` is never closed [2]), and they
survive until IOManager/TaskExecutor shutdown deletes the managed dir —
effectively never on a long-lived pooled TM. Recovery-failure loops then
accumulate GB-scale spill.
Paths that spill but never drive cleanup:
1. `readInputData` throws after ≥1 file spilled — `fetchChannelState` swallows
it and returns `Optional.empty()`, the produced state is dropped [3].
2. the `requestPartitions` mail throws/drops → `buildDrainer` never runs.
3. the trigger-install mail throws/drops → `drain` never runs.
4. `thenRunAsync(drain, channelIOExecutor)` is rejected after the executor shut
down → `.exceptionally` only logs [4].
Fix: give the produced state/drainer a single owner that closes it on any abort
(e.g. register with `cancelables`). Separately, `deleteAllFiles()` never
removes the empty `baseDir`, and spilling pins `spillTmpDirectories[0]` instead
of striping like `FileChannelManagerImpl`.
[1]
[https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/channel/FetchedChannelState.java#L101-L130]
[2]
[https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java#L951-L1000]
[3]
[https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java#L1078-L1094]
[4]
[https://github.com/apache/flink/blob/76e774e04ff84ea305714153c5f4790fff98437c/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java#L960-L1000]
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
