Skip to content

[FLINK-39923][state] Fix RocksDB Statistics native memory leak #28489

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
leekeiabstraction wants to merge 1 commit into
base: master
Choose a base branch
from
+26 −10
Open

[FLINK-39923][state] Fix RocksDB Statistics native memory leak #28489

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions ...kend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBNativeMetricMonitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
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;
Expand Down Expand Up @@ -107,7 +108,7 @@ void registerColumnFamily(String columnFamilyName, ColumnFamilyHandle handle) {

/** 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 {
Expand All @@ -126,11 +127,11 @@ private void setProperty(RocksDBNativePropertyMetricView metricView) {
}

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));
}
}
Expand All @@ -140,6 +141,8 @@ private void setStatistics(RocksDBNativeStatisticsMetricView metricView) {
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;
}
}
Expand Down
25 changes: 19 additions & 6 deletions ...tebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/restore/RocksDBHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.rocksdb.DBOptions;
import org.rocksdb.ExportImportFilesMetaData;
import org.rocksdb.RocksDB;
import org.rocksdb.Statistics;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -147,12 +150,16 @@ private void loadDb() throws IOException {
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(
Expand Down Expand Up @@ -306,7 +313,13 @@ public DBOptions getDbOptions() {
@Override
public void close() throws Exception {
IOUtils.closeQuietly(defaultColumnFamilyHandle);
IOUtils.closeQuietly(nativeMetricMonitor);
if (nativeMetricMonitor != null) {
// Monitor owns the statistics wrapper.
IOUtils.closeQuietly(nativeMetricMonitor);
} else {
Comment on lines +316 to +319

@snuyanzin snuyanzin Jun 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

closeQuietly has it's own internal non null check
given that, can it be simplified to just

        IOUtils.closeQuietly(nativeMetricMonitor);
        IOUtils.closeQuietly(statistics);

?

@rkhachatryan rkhachatryan Jun 19, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IIUC that might cause double close of statistics (here and in nativeMetricMonitor).

@snuyanzin snuyanzin Jun 20, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will it be an issue (double closing), asking since closeQuietly should catch all the exceptions?
Or we should keep unclosed in some cases?

anyway it is just a nit pick, so feel free to merge

// 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()));
Expand Down