[
https://issues.apache.org/jira/browse/SPARK-57183?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Iván Morales updated SPARK-57183:
---------------------------------
Description:
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.
was:
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.
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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: [email protected]
For additional commands, e-mail: [email protected]
