[jira] [Commented] (FLINK-40037) Local RocksDB working directory leaked (with all file handles held open) when an async-state-v2 task is redeployed onto the same TaskManager after a job restart

["Kartikey Pant (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-40037?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18093245#comment-18093245 ] 

Kartikey Pant commented on FLINK-40037:
---------------------------------------

Thanks [~gliter] for the write-up, the repro matches what I see in the code.

[~Zakelly] [~Yanfei Lei] could one of you assign this to me? I'd like to take it.

{{AsyncKeyedStateBackendAdaptor.dispose()}} and {{close()}} are empty
({{{}AsyncKeyedStateBackendAdaptor.java:133,143{}}}). When {{StreamTaskStateInitializerImpl}} wraps a v1
{{RocksDBKeyedStateBackend}} in the adaptor for async-state-v2 operators, the wrapped backend's {{dispose()}} is
never called, so {{cleanInstanceBasePath()}} doesn't run and the local dir leaks on every redeploy.

Delegating both methods to the wrapped backend fixes it. Will open a PR.

> Local RocksDB working directory leaked (with all file handles held open) when an async-state-v2 task is redeployed onto the same TaskManager after a job restart
> ----------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-40037
>                 URL: https://issues.apache.org/jira/browse/FLINK-40037
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / State Backends
>    Affects Versions: 2.2.1
>            Reporter: Grzegorz Liter
>            Priority: Major
>         Attachments: image-2026-07-01-11-04-38-426.png
>
>
> When using the State V2 / async state API (AsyncKeyedProcessOperator, StateFuture) on top of the classic embedded RocksDB state backend (state.backend: rocksdb, not ForSt), a full job restart that redeploys a subtask onto a TaskManager it was already running on causes the previous attempt's local RocksDB working directory to leak permanently — both on disk and as open file descriptors — instead of being cleaned up.
> Each task attempt creates a local working directory of the form:
> <taskmanager-tmp-dir>/tmp/job_<jobId>_op_<operatorId>__<subtaskIndex>_<parallelism>__uuid_<uuid>/db
> When the job restarts and the subtask is rescheduled onto the same TaskManager, a new UUID directory is created for the new attempt, but the old directory is never removed, and — more importantly — the old AsyncKeyedStateBackendAdaptor/RocksDBKeyedStateBackend instance backing it is apparently never disposed: every SST/data file in the old directory remains open by the TaskManager JVM indefinitely (confirmed via /proc/<pid>/fd), with no decrease over time. Because state.backend.rocksdb.files.open defaults to -1 (RocksDB keeps every table file open for the life of the DB object, no LRU eviction), none of these leaked handles are ever released.
> This has two compounding effects:
> 1. Disk usage roughly doubles (or worse, with repeated restarts) on every TaskManager pod that survives a job restart but gets a subtask redeployed onto it, since the old directory's disk blocks can't be reclaimed even by manually deleting the files (see reproduction step 6) — Linux won't free blocks for a file while any process still holds it open.
> 2. Because disk fills up faster on surviving TMs, this creates a cascading failure: one TM running low on disk → job-wide restart → every surviving TM leaks another copy of its local state → disk fills faster next time → more restarts.
>  
> !image-2026-07-01-11-04-46-408.png!



--
This message was sent by Atlassian Jira
(v8.20.10#820010)