[jira] [Assigned] (FLINK-40952) ForSt: rescale restore never closes the temporary DB's DBOptions copy, leaking the slot-shared block cache

["Zakelly Lan (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40952?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Zakelly Lan reassigned FLINK-40952:
-----------------------------------

    Assignee: Yordan Pavlov

> ForSt: rescale restore never closes the temporary DB's DBOptions copy, leaking the slot-shared block cache
> ----------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-40952
>                 URL: https://issues.apache.org/jira/browse/FLINK-40952
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / State Backends
>    Affects Versions: 2.0.0, 2.1.0, 2.2.0, 2.3.0
>            Reporter: Yordan Pavlov
>            Assignee: Yordan Pavlov
>            Priority: Major
>              Labels: pull-request-available
>         Attachments: DbOptionsCopyLeak.java, FLINK-forst-temp-dboptions-master.diff, ForStIncrementalRestoreOperation.java, ScaleInLeakCheck.java, scale-in-fixed.log, scale-in-stock.log
>
>
> h3. Problem
> {{ForStIncrementalRestoreOperation#restoreTempDBInstance}} opens every temporary restore DB with a native copy of the backend's DB options, and nothing ever closes it:
> {code:java}
> // do not relocate log file for temp db
> DBOptions dbOptions = new DBOptions(this.forstHandle.getDbOptions());
> dbOptions.setDbLogDir("");
> RocksDB restoreDb = ForStOperationUtils.openDB(..., dbOptions);
> return new RestoredDBInstance(restoreDb, ...); // dbOptions is not passed on and not closed
> {code}
> {{RestoredDBInstance#close()}} closes the DB, its column family handles/options and its {{ReadOptions}}, but not this {{DBOptions}}.
> With managed memory ({{state.backend.forst.memory.managed: true}}, the default), the copied native {{DBOptions}} shares the slot's {{WriteBufferManager}} (a {{shared_ptr}} in the C++ {{DBOptions}}), and the WBM holds the slot-shared block cache, because memtable memory is charged to the cache. When the backend is later disposed, all Java handles ({{RocksDB}}, {{LRUCache}}, {{WriteBufferManager}}, {{DBOptions}}) are closed, but the leaked copy keeps the native WBM, and through it the whole block cache with its cached blocks, alive until the TaskManager JVM exits.
> Temporary DBs are used for multi-handle restores (every scale-in and some scale-outs), so each such restore pins the block cache of the new backend. That memory leaks at the next restart on the same TaskManager: rescale, failover, or an in-place restart by the adaptive scheduler or autoscaler. Each leak is up to one slot's block cache capacity. A few rescales or restart attempts on a long-lived TM can OOM-kill the container.
> * Not affected: unmanaged ForSt memory (no WBM, the copy pins only a few KB), same-parallelism failovers (single-handle restore, no temp DB), and the RocksDB backend ({{RocksDBIncrementalRestoreOperation}} does not copy the options).
> h3. Evidence
> *Production* (ForSt + async state V2, Flink 2.3.0, one TM with 3 slots and ~0.5 GB block cache per slot, job restarted in place via {{PUT /jobs/<id>/resource-requirements}}). "Unexplained native" = native anon RSS outside the heap, minus the live ForSt {{block-cache-usage}} and memtables:
> ||Restart||Backends disposed||Their cache when disposed||Unexplained native||Step||
> |baseline p=1|-|-|3,130 MiB| |
> |p=1 → 3|1 (restored by a scale-in)|full, ~420 MiB|3,576 MiB|*+446*|
> |p=3 → 1|3|young, ~100–200 MiB|3,604 MiB|+28|
> |p=1 → 3|1 (restored by a scale-in)|full, ~413 MiB|4,040 MiB|*+436*|
> |p=3 → 1|3|young, ~100–200 MiB|4,087 MiB|+47|
> * jemalloc stats on the live TM: the retained memory is live allocations, not pages kept by the allocator (allocated 2.76 GiB, resident 3.16 GiB, dirty 34 MiB).
> * A heap dump shows that every Java-side object of the disposed backends is closed ({{owningHandle_ = false}}), so the memory is held from the native side.
> * With {{state.backend.forst.memory.managed: false}} (private caches, no WBM), the same restart freed all of the disposed backends' ForSt memory (+1 MiB).
> *JNI-only reproduction* (forstjni 0.1.8, jemalloc with {{dirty_decay_ms:0,muzzy_decay_ms:0}}): a shared {{LRUCache}} + {{WriteBufferManager(cache)}} + {{DBOptions}} with the WBM, cache filled to ~308 MiB, then everything closed. Retained RSS: 94 MiB with no copy, *422 MiB with an unclosed {{new DBOptions(options)}} copy*, 90 MiB when the copy is closed too.
> *MiniCluster reproduction* (attached {{ScaleInLeakCheck.java}}): a ForSt async-state job with 400k keys, a 2-slot TM with 1 GB managed memory, restored alternately at p=2 and p=1 from native savepoints in one JVM. RSS after each run's MiniCluster has shut down, stock flink-dist 2.3.0 vs. the same with the fix:
> ||Run||Stock RSS||Step||Fixed RSS||Step||
> |base|2,150 MiB||2,150 MiB||
> |0: p=2, fresh|2,307 MiB|+157|2,308 MiB|+158|
> |1: p=1 (scale-in)|2,713 MiB|*+406*|2,354 MiB|+46|
> |2: p=2|2,759 MiB|+46|2,403 MiB|+49|
> |3: p=1 (scale-in)|3,182 MiB|*+423*|2,441 MiB|+38|
> |4: p=2|3,214 MiB|+32|2,468 MiB|+27|
> |5: p=1 (scale-in)|3,632 MiB|*+418*|2,504 MiB|+36|
> |total|| *+1,482 MiB* || +354 MiB|
> Stock leaks about one slot's block cache (~410 MiB) on every scale-in. With the fix, every run adds 27–49 MiB regardless of direction, with no scale-in pattern (JVM and allocator warm-up). Run with {{-Xms2g -Xmx2g -XX:+AlwaysPreTouch}} and {{LD_PRELOAD=libjemalloc.so.2 MALLOC_CONF=dirty_decay_ms:0,muzzy_decay_ms:0}}, so that RSS reflects live native allocations.
> h3. Proposed fix
> Pass the copy to {{RestoredDBInstance}}, close it in {{close()}} right after the DB, and close it in {{restoreTempDBInstance}} if {{openDB}} fails. The copy is created per temporary DB and used only to open it, so closing it drops only this copy's references and doesn't affect the main DB's options, WBM or cache. PR to follow.
> We are deploying this patch to production on top of 2.3.0.



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