You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Rui Fan (Jira)" <ji...@apache.org> on 2026/09/01 15:06:00 UTC
[jira] [Updated] (FLINK-40520) Recovered channel-state spill files leak when recovery aborts between fetch and drain
[ https://issues.apache.org/jira/browse/FLINK-40520?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Rui Fan updated FLINK-40520:
----------------------------
Description:
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`).
[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]
was:
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]
> 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
> Priority: Major
>
> 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`).
> [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)