[jira] [Resolved] (SPARK-57183) RocksDB state store leaks native memory when LRUCache is not closed in unbounded mode

["Anish Shrigondekar (Jira)" <ji...@apache.org>]

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

Anish Shrigondekar resolved SPARK-57183.
----------------------------------------
    Fix Version/s: 5.0.0
       Resolution: Fixed

Issue resolved by pull request 56234
[https://github.com/apache/spark/pull/56234]

> RocksDB state store leaks native memory when LRUCache is not closed in unbounded mode
> -------------------------------------------------------------------------------------
>
>                 Key: SPARK-57183
>                 URL: https://issues.apache.org/jira/browse/SPARK-57183
>             Project: Spark
>          Issue Type: Bug
>          Components: Structured Streaming
>    Affects Versions: 4.2.0
>            Reporter: Iván Morales
>            Assignee: Iván Morales
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: 5.0.0
>
>
> In unbounded memory mode (the default, {{{}boundedMemoryUsage = false{}}}), Spark's RocksDB state store wrapper creates a new {{LRUCache}} per instance in {{RocksDBMemoryManager }}but never calls {{lruCache.close()}} in {{{}RocksDB.close(){}}}. The Java {{LRUCache}} wrapper holds a C++ {{{}shared_ptr<Cache>{}}}, preventing native memory from being freed until the JVM GC finalizes the wrapper. Under low heap pressure (e.g., a 4 GB JVM running streaming tests), GC rarely runs, so unclosed caches accumulate. Over a CI run with ~49 RocksDB-heavy test suites this accumulates 4-8 GB of native memory and causes OOM kills.
> *Affected file*
> {{sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala}}
> *Related:* SPARK-56523 (Statistics native memory leak – already fixed). This is a separate, unfixed issue.
> h2. Root cause
> In {{{}RocksDBMemoryManager.scala{}}}, unbounded mode creates a new cache per instance:
> {code:scala}
> } else {
>   (null, new LRUCache(conf.blockCacheSizeMB * 1024 * 1024))  // new instance per RocksDB
> }
> {code}
> But {{RocksDB.close()}} never closes it:
> {code:scala}
> def close(): Unit = {
>   closeDB()
>   readOptions.close()
>   writeOptions.close()
>   flushOptions.close()
>   nativeStats.close()    // added by SPARK-56523 fix
>   rocksDbOptions.close()
>   dbLogger.close()
>   // lruCache.close()  <-- MISSING
> }
> {code}
> In bounded mode ({{{}boundedMemoryUsage = true{}}}) the cache is a shared singleton in {{RocksDBMemoryManager}} and must not be closed per instance. In unbounded mode (the default) each instance owns its cache and must close it.
> h2. Fix
> Add the following in {{{}RocksDB.close(){}}}, after {{{}dbLogger.close(){}}}:
> {code:scala}
> if (!conf.boundedMemoryUsage && lruCache != null) {
>   lruCache.close()
> }
> {code}
> h2. Evidence
> Standalone reproducer tool: [https://github.com/kete1987/rocksdb-leak-tool]
> Results (60 iterations, --no-gc, --cache-mb 8, Windows 11 / Java 11.0.28):
> ||Mode||Total growth||Per-iter (post-warmup)||Pattern||
> |leak|+578 MB|~8.5 MB|Linear|
> |fixed|+128 MB|~0 KB|Flat|
> The ~8.5 MB/iter matches exactly {{blockCacheSizeMB = 8}} (Spark default), confirming {{LRUCache}} as the sole leak source.



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

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org