[
https://issues.apache.org/jira/browse/FLINK-40520?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40520:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Labels: pull-request-available
>
> 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)
