[jira] [Created] (FLINK-40520) Recovered channel-state spill files leak when recovery aborts between fetch and drain

["Rui Fan (Jira)" <ji...@apache.org>]

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)