KAFKA-20287 : Fix CF handle leaks by muralibasani · Pull Request #21751 · apache/kafka

muralibasani · 2026-03-13T19:13:01Z

RocksDBStore.java — openRocksDB() base method: Added finally block
after RocksDB.open() to close all CF handles and db if
createColumnFamilies() or mergeColumnFamilyHandleLists() fails.
RocksDBStore.java — openDB(): Added finally block with
closeNativeResources() to close all partially-initialized native
resources if openRocksDB() or cfAccessor.open() fails.
RocksDBTimestampedStore.java: Changed RocksIterator from manual
close() to try-with-resources, and added finally block to close all CF
handles if an exception occurs before the accessor takes ownership.
RocksDBMigratingSessionStoreWithHeaders.java: Same as above —
try-with-resources for RocksIterator and finally block for CF handle
cleanup on failure.
RocksDBTimestampedStoreWithHeaders.java — openFromDefaultStore():
Added finally block to close all CF handles if an exception occurs
before the accessor takes ownership.
RocksDBTimestampedStoreWithHeaders.java — openFromTimestampedStore():
Replaced manual per-handle close() calls (which missed
columnFamilies.get(3)) with a single finally block that loops over all
handles on failure.

Reviewers: Matthias J. Sax matthias@confluent.io, Alieh Saeedi
asaeedi@confluent.io

muralibasani · 2026-03-13T19:36:36Z

-            throw new ProcessorStateException(fatalMessage, fatal);
-        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error opening store " + name, e);
+            openRocksDB(dbOptions, columnFamilyOptions);


This method internally calls createColumnFamilies() or mergeColumnFamilyHandleLists() which might fail

I am wondering if we should do the necessary exception-handling inside openRocksDB instead?

I see openRocksDB() already handles its own cleanup (CF handles + db), but the closeNativeResources() in openDB() covers a different failure scope that openRocksDB() can't handle.

These 6 native resources (userSpecifiedOptions, cache, filter, wOptions, fOptions, statistics) are all created in openDB() before openRocksDB() is ever called and openRocksDB() doesn't own them and has no references to clean them up.

If openRocksDB() itself fails, it will cleanup everything which is in finally block.

If openRocksDB() does not fail but cfAccessor.open() fails — now db is open. Hence closeNativeResources handles full teardown.

Other option is to move everything into openRocksDB(), but this would change the method signature across all overrides.

muralibasani · 2026-03-13T19:44:55Z

+            success = true;
+        } finally {
+            if (!success) {
+                closeNativeResources();


if openRocksDB() or cfAccessor.open() fail

muralibasani · 2026-03-13T19:49:30Z

-            noTimestampColumnFamily.close();
+        boolean success = false;
+        try {
+            try (final RocksIterator noTimestampsIter = db.newIterator(noTimestampColumnFamily)) {


updating with try-resources

muralibasani · 2026-03-13T19:50:04Z

+        boolean success = false;
+        try {
+            // Check if default CF has data (plain store upgrade)
+            try (final RocksIterator defaultIter = db.newIterator(defaultCf)) {


also updating with try-with resources

muralibasani · 2026-03-13T19:51:31Z

+        boolean success = false;
+        try {
+            // verify and close empty Default ColumnFamily
+            try (final RocksIterator defaultIter = db.newIterator(columnFamilies.get(0))) {


updating here as well in the same way try with resources

muralibasani · 2026-03-13T19:55:06Z

-            noHeadersColumnFamily.close();
+        boolean success = false;
+        try {
+            try (final RocksIterator noHeadersIter = db.newIterator(noHeadersColumnFamily)) {


updating here with try-with resources

muralibasani · 2026-03-16T15:20:17Z

@mjsax do you think this pr addresses the cf handle leaks ? Possible to review ?

mjsax

Thanks for the PR. Made a first pass. -- In the meantime, we added RocksDBMigratingWindowStoreWithHeaders that we should include in this PR.

mjsax · 2026-03-20T00:42:25Z

+     * Used only by the error cleanup path in {@link #openDB} where some resources
+     * may not have been initialized yet.
+     */
+    private void closeNativeResources() {


We do already have code in close() which closes all these resources -- are these new helper necessary? I would believe that the runtime ensures that we call RocksDBStore.close() if init() (which calls openDB()) fails.

In close method, we are checking isOpen method which in turn checks open.get()

this would wait for openDB/openRocksDb() and if these fail, it would return false.

this 'false' does not close all resources in close() method.

mjsax · 2026-03-20T00:44:54Z

-            throw new ProcessorStateException(fatalMessage, fatal);
-        } catch (final RocksDBException e) {
-            throw new ProcessorStateException("Error opening store " + name, e);
+            openRocksDB(dbOptions, columnFamilyOptions);


I am wondering if we should do the necessary exception-handling inside openRocksDB instead?

mjsax · 2026-03-20T00:49:47Z

+                    mergeColumnFamilyHandleLists(existingColumnFamilies, createdColumnFamilies, allDescriptors);
+                openSuccess = true;
+                return result;
+            } finally {


Why are we not use catch(Exception e) -- this allows us to drop openSuccess flag?

Agree with removing success flag. Updated with catch block.

mjsax · 2026-03-20T00:50:37Z

+                    for (final ColumnFamilyHandle handle : createdColumnFamilies) {
+                        handle.close();
+                    }
+                    db.close();


I don't think we need to close the db -- this should happen in RocksDBStore.close() already?

Yes, db.close() is not required again. Removed.

mjsax · 2026-03-20T00:51:26Z

                    .collect(Collectors.toList());
            final List<ColumnFamilyHandle> existingColumnFamilies = new ArrayList<>(existingDescriptors.size());
+            final List<ColumnFamilyHandle> createdColumnFamilies = new ArrayList<>();
            db = RocksDB.open(dbOptions, absolutePath, existingDescriptors, existingColumnFamilies);


I think this call should also go inside the try-catch block?

Yes. The RocksDB.open() call is now inside the try-catch block, so if createColumnFamilies() or mergeColumnFamilyHandleLists() throws, the existingColumnFamilies handles populated by RocksDB.open() are properly cleaned up.

mjsax · 2026-03-20T00:53:22Z

+            boolean openSuccess = false;
+            try {
+                createdColumnFamilies.addAll(db.createColumnFamilies(toCreate));
+                final List<ColumnFamilyHandle> result =


Suggested change

final List<ColumnFamilyHandle> result =

final List<ColumnFamilyHandle> allColumnFamilies =

Now returning it inline, instead of new var

mjsax · 2026-03-20T00:55:09Z

+                }
+            }
+            success = true;
+        } finally {


Why not use catch(Exception) and get rid of success flag?

Done. But also had to replace Exception with RunTimeException (spotbugs)

mjsax · 2026-03-20T00:55:35Z

+                }
+            }
+            success = true;
+        } finally {


mjsax · 2026-03-20T00:56:16Z

+                }
+            }
+            success = true;
+        } finally {


Applied the above change here too.

mjsax · 2026-03-20T00:57:38Z

+                }
+            }
+            success = true;
+        } finally {


We should use catch(Exception) instead

aliehsaeedii

Thanks @muralibasani. Good catch.
Just needs a rebase (and former comments from Matthias)

muralibasani · 2026-03-20T16:10:24Z

@mjsax @aliehsaeedii Updated PR based on comments.
Pls take another look.

muralibasani · 2026-03-20T16:57:26Z

-                noHeadersColumnFamily.close();
+        try {
+            // Check if DEFAULT CF has data (upgrade from old format without headers)
+            try (final RocksIterator noHeadersIter = db.newIterator(noHeadersColumnFamily)) {


moving newIterator() into try-resources

mjsax

Thanks. A few more cosmetics. Overall LGTM.

mjsax · 2026-03-20T22:14:25Z

        setupStatistics(configs, dbOptions);
-        openRocksDB(dbOptions, columnFamilyOptions);
-        dbAccessor = new DirectDBAccessor(db, fOptions, wOptions);
+        boolean success = false;


Seems you missed this one for removal?

Ah yes. Done.

mjsax · 2026-03-20T22:14:46Z

+                throw new ProcessorStateException("Error opening store " + name, e);
+            }
+            success = true;
+        } finally {


Should we use catch() instead?

Agree, removed.

mjsax · 2026-03-20T22:16:32Z

+                        withHeadersColumnFamily,
+                        HeadersBytesStore::convertToHeaderFormat,
+                        this,
+                            open


Suggested change

open

open

mjsax · 2026-03-20T22:18:53Z

+                        noTimestampColumnFamily,
+                        withTimestampColumnFamily,
+                        TimestampedBytesStore::convertToTimestampedFormat,
+                        this, open


Suggested change

this, open

this,

open

mjsax · 2026-03-20T22:19:28Z

+                        headersCf,
+                        HeadersBytesStore::convertFromPlainToHeaderFormat,
+                        this,
+                            open


Suggested change

open

open

mjsax · 2026-03-20T22:21:12Z

-            log.info("Opening store {} in regular mode", name);
-            cfAccessor = new SingleColumnFamilyAccessor(offsetsCf, withHeadersColumnFamily);
-            noHeadersColumnFamily.close();
+        try {


Why do we need to nest two try-catch block? With success flag begin remove, isn't one sufficient? (Also elsewhere.)

Indeed not needed. Removed from others too.

muralibasani · 2026-03-21T17:34:08Z

Thanks. A few more cosmetics. Overall LGTM.

Thank you @mjsax . Pls take another look.

- RocksDBStore.java — openRocksDB() base method: Added finally block after RocksDB.open() to close all CF handles and db if createColumnFamilies() or mergeColumnFamilyHandleLists() fails. - RocksDBStore.java — openDB(): Added finally block with closeNativeResources() to close all partially-initialized native resources if openRocksDB() or cfAccessor.open() fails. - RocksDBTimestampedStore.java: Changed RocksIterator from manual close() to try-with-resources, and added finally block to close all CF handles if an exception occurs before the accessor takes ownership. - RocksDBMigratingSessionStoreWithHeaders.java: Same as above — try-with-resources for RocksIterator and finally block for CF handle cleanup on failure. - RocksDBTimestampedStoreWithHeaders.java — openFromDefaultStore(): Added finally block to close all CF handles if an exception occurs before the accessor takes ownership. - RocksDBTimestampedStoreWithHeaders.java — openFromTimestampedStore(): Replaced manual per-handle close() calls (which missed columnFamilies.get(3)) with a single finally block that loops over all handles on failure. Reviewers: Matthias J. Sax <matthias@confluent.io>, Alieh Saeedi <asaeedi@confluent.io>

mjsax · 2026-03-27T22:46:33Z

Thanks for the PR!

Merged to trunk and cherry-picked to 4.3 branch.

- RocksDBStore.java — openRocksDB() base method: Added finally block after RocksDB.open() to close all CF handles and db if createColumnFamilies() or mergeColumnFamilyHandleLists() fails. - RocksDBStore.java — openDB(): Added finally block with closeNativeResources() to close all partially-initialized native resources if openRocksDB() or cfAccessor.open() fails. - RocksDBTimestampedStore.java: Changed RocksIterator from manual close() to try-with-resources, and added finally block to close all CF handles if an exception occurs before the accessor takes ownership. - RocksDBMigratingSessionStoreWithHeaders.java: Same as above — try-with-resources for RocksIterator and finally block for CF handle cleanup on failure. - RocksDBTimestampedStoreWithHeaders.java — openFromDefaultStore(): Added finally block to close all CF handles if an exception occurs before the accessor takes ownership. - RocksDBTimestampedStoreWithHeaders.java — openFromTimestampedStore(): Replaced manual per-handle close() calls (which missed columnFamilies.get(3)) with a single finally block that loops over all handles on failure. Reviewers: Matthias J. Sax <matthias@confluent.io>, Alieh Saeedi <asaeedi@confluent.io>

github-actions bot added triage PRs from the community streams labels Mar 13, 2026

muralibasani commented Mar 13, 2026

View reviewed changes

muralibasani changed the title ~~Fix CF handle leaks~~ KAFKA-20287 : Fix CF handle leaks Mar 13, 2026

muralibasani marked this pull request as ready for review March 13, 2026 19:56

github-actions bot removed the triage PRs from the community label Mar 14, 2026

mjsax reviewed Mar 20, 2026

View reviewed changes

mjsax added kip Requires or implements a KIP ci-approved labels Mar 20, 2026

aliehsaeedii approved these changes Mar 20, 2026

View reviewed changes

muralibasani force-pushed the fix-cf-leaks branch from 0587c17 to 22b1dd2 Compare March 20, 2026 15:16

muralibasani commented Mar 20, 2026

View reviewed changes

mjsax reviewed Mar 20, 2026

View reviewed changes

muralibasani added 3 commits March 21, 2026 18:13

Fix CF handle leaks

9ac3fe9

Fix from reviews

1c031a5

Fix from reviews

230723a

muralibasani force-pushed the fix-cf-leaks branch from 2f24bc1 to 230723a Compare March 21, 2026 17:13

Remove double try blocks, indentation issues

76bd45d

muralibasani requested a review from mjsax March 21, 2026 17:32

mjsax approved these changes Mar 27, 2026

View reviewed changes

mjsax merged commit 5e509dc into apache:trunk Mar 27, 2026
26 checks passed

	final List<ColumnFamilyHandle> result =
	final List<ColumnFamilyHandle> allColumnFamilies =

Conversation

muralibasani commented Mar 13, 2026 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 16, 2026

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 20, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

muralibasani commented Mar 13, 2026 •

edited by github-actions bot

Loading

aliehsaeedii left a comment •

edited

Loading

muralibasani commented Mar 21, 2026 •

edited

Loading