This is an automated email from the ASF dual-hosted git repository.
rkhachatryan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 3e3e270c98d [FLINK-39923][state] Fix RocksDB Statistics native memory
leak (#28489)
3e3e270c98d is described below
commit 3e3e270c98d9205d1eac3d4baea3a20c373805a2
Author: Keith Lee <[email protected]>
AuthorDate: Tue Jul 7 23:01:08 2026 +0100
[FLINK-39923][state] Fix RocksDB Statistics native memory leak (#28489)
Close the JNI Statistics wrapper in `RocksDBNativeMetricMonitor.close()`
and on partial init in `RocksDBHandle.loadDb()`.
---
.../state/rocksdb/RocksDBNativeMetricMonitor.java | 11 ++++++----
.../flink/state/rocksdb/restore/RocksDBHandle.java | 25 ++++++++++++++++------
2 files changed, 26 insertions(+), 10 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricMonitor.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricMonitor.java
index e51ec32c7da..83a885a4f47 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricMonitor.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricMonitor.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.View;
+import org.apache.flink.util.IOUtils;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDB;
@@ -107,7 +108,7 @@ public class RocksDBNativeMetricMonitor implements
Closeable {
/** Updates the value of metricView if the reference is still valid. */
private void setProperty(RocksDBNativePropertyMetricView metricView) {
- if (metricView.isClosed()) {
+ if (metricView.isClosed() || rocksDB == null) {
return;
}
try {
@@ -126,11 +127,11 @@ public class RocksDBNativeMetricMonitor implements
Closeable {
}
private void setStatistics(RocksDBNativeStatisticsMetricView metricView) {
- if (metricView.isClosed()) {
+ if (metricView.isClosed() || statistics == null) {
return;
}
- if (statistics != null) {
- synchronized (lock) {
+ synchronized (lock) {
+ if (statistics != null) {
metricView.setValue(statistics.getTickerCount(metricView.tickerType));
}
}
@@ -140,6 +141,8 @@ public class RocksDBNativeMetricMonitor implements
Closeable {
public void close() {
synchronized (lock) {
rocksDB = null;
+ // Wrapper holds a JNI shared_ptr that leaks without explicit
close. See FLINK-39923.
+ IOUtils.closeQuietly(statistics);
statistics = null;
}
}
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java
index c233922d9d2..5cbd60b3a15 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java
@@ -37,6 +37,7 @@ import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.DBOptions;
import org.rocksdb.ExportImportFilesMetaData;
import org.rocksdb.RocksDB;
+import org.rocksdb.Statistics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -90,6 +91,8 @@ class RocksDBHandle implements AutoCloseable {
private RocksDB db;
private ColumnFamilyHandle defaultColumnFamilyHandle;
private RocksDBNativeMetricMonitor nativeMetricMonitor;
+ // Released in close() for the partial-init case; on success the monitor
closes it.
+ private Statistics statistics;
private final Long writeBufferManagerCapacity;
protected RocksDBHandle(
@@ -147,12 +150,16 @@ class RocksDBHandle implements AutoCloseable {
dbOptions);
// remove the default column family which is located at the first index
defaultColumnFamilyHandle = columnFamilyHandles.remove(0);
- // init native metrics monitor if configured
+
+ if (!nativeMetricOptions.isEnabled()) {
+ return;
+ }
+ // dbOptions.statistics() returns a new Java wrapper around a fresh
shared_ptr<Statistics>
+ // aliasing the existing native StatisticsImpl. The original wrapper
is closed by
+ // RocksDBResourceContainer (via dbOptions); this one must also be
closed. See FLINK-39923.
+ statistics = dbOptions.statistics();
nativeMetricMonitor =
- nativeMetricOptions.isEnabled()
- ? new RocksDBNativeMetricMonitor(
- nativeMetricOptions, metricGroup, db,
dbOptions.statistics())
- : null;
+ new RocksDBNativeMetricMonitor(nativeMetricOptions,
metricGroup, db, statistics);
}
RocksDbKvStateInfo getOrRegisterStateColumnFamilyHandle(
@@ -306,7 +313,13 @@ class RocksDBHandle implements AutoCloseable {
@Override
public void close() throws Exception {
IOUtils.closeQuietly(defaultColumnFamilyHandle);
- IOUtils.closeQuietly(nativeMetricMonitor);
+ if (nativeMetricMonitor != null) {
+ // Monitor owns the statistics wrapper.
+ IOUtils.closeQuietly(nativeMetricMonitor);
+ } else {
+ // Monitor construction never completed; release the wrapper
directly.
+ IOUtils.closeQuietly(statistics);
+ }
IOUtils.closeQuietly(db);
// Making sure the already created column family options will be closed
columnFamilyDescriptors.forEach((cfd) ->
IOUtils.closeQuietly(cfd.getOptions()));
