You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2026/06/08 13:39:54 UTC
(kafka) branch 4.3 updated: KAFKA-20616: Close offsets-CF ColumnFamilyOptions and ensure CF handles are releas… (#22378)
This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new 66a673a5961 KAFKA-20616: Close offsets-CF ColumnFamilyOptions and ensure CF handles are releas… (#22378)
66a673a5961 is described below
commit 66a673a5961be7540ce3f83a53f1000ee5d3da37
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Mon Jun 8 09:29:35 2026 -0400
KAFKA-20616: Close offsets-CF ColumnFamilyOptions and ensure CF handles are releas… (#22378)
Fixes two close-path native memory leaks in `RocksDBStore`.
### Issue 1 — leaked `ColumnFamilyOptions`
`createOffsetsCFOptions()` returned a `new ColumnFamilyOptions()` that
was passed to a `ColumnFamilyDescriptor` and then dropped — never stored
on a field, never closed. Each construction auto-allocates a default
`BlockBasedTableFactory` + 8 MB `LRUCache` on the JNI side;
segmented/windowed stores amplify the leak per segment per task.
**Fix:** store the options on a `RocksDBStore.offsetsCfOptions` field
and close it in `close()` / `closeNativeResources()`.
### Issue 2 — CF handles leak on close-path exceptions (KIP-1035)
`AbstractColumnFamilyAccessor.close()` writes a `closedState` marker
to the offsets CF. If that write throws (EOSv2 fencing, unclean shutdown
— already noted in the existing code comment), the subsequent
`offsetColumnFamilyHandle.close()` was skipped.
`SingleColumnFamilyAccessor.close()` and
`DualColumnFamilyAccessor.close()` had the same non-`finally` ordering,
so the data / old / new CF handles leaked too.
`RocksDBStore.close()` swallows the resulting `RocksDBException`, making
the leak silent.
**Fix:** `try/finally` in all three `close()` methods so every
`ColumnFamilyHandle.close()` runs even when an earlier write or
super-close throws
Reviewers: Matthias J. Sax <ma...@confluent.io>, Lucas Brutschy
<lb...@confluent.io>
---
.../internals/AbstractColumnFamilyAccessor.java | 18 ++-
.../state/internals/DualColumnFamilyAccessor.java | 22 ++-
.../RocksDBMigratingSessionStoreWithHeaders.java | 2 +-
.../RocksDBMigratingWindowStoreWithHeaders.java | 2 +-
.../streams/state/internals/RocksDBStore.java | 48 +++++--
.../state/internals/RocksDBTimestampedStore.java | 2 +-
.../RocksDBTimestampedStoreWithHeaders.java | 4 +-
...ocksDBMigratingSessionStoreWithHeadersTest.java | 69 +++++++++
...gratingWindowStoreWithHeadersCloseLeakTest.java | 156 +++++++++++++++++++++
.../state/internals/RocksDBStoreCloseLeakTest.java | 139 ++++++++++++++++++
.../internals/RocksDBTimestampedStoreTest.java | 36 +++++
.../RocksDBTimestampedStoreWithHeadersTest.java | 46 ++++++
.../internals/ThrowingOnOffsetsPutDBAccessor.java | 98 +++++++++++++
13 files changed, 615 insertions(+), 27 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
index 004c824cc3e..d0c1ff8ea56 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractColumnFamilyAccessor.java
@@ -98,11 +98,16 @@ abstract class AbstractColumnFamilyAccessor implements RocksDBStore.ColumnFamily
public void close(final RocksDBStore.DBAccessor accessor) throws RocksDBException {
// Only persist the closed state if the store was previously open.
// After an unclean shutdown, RocksDB may still be running background recovery,
- // causing accessor.put() to block.
- if (storeOpen.compareAndSet(true, false)) {
- accessor.put(offsetColumnFamilyHandle, statusKey, closedState);
+ // causing accessor.put() to block. The put can also throw if RocksDB is in a
+ // failed state (e.g. during an EOSv2 fencing cascade); the handle close must
+ // still happen, otherwise the native ColumnFamilyHandle leaks every cycle.
+ try {
+ if (storeOpen.compareAndSet(true, false)) {
+ accessor.put(offsetColumnFamilyHandle, statusKey, closedState);
+ }
+ } finally {
+ offsetColumnFamilyHandle.close();
}
- offsetColumnFamilyHandle.close();
}
@Override
@@ -115,6 +120,11 @@ abstract class AbstractColumnFamilyAccessor implements RocksDBStore.ColumnFamily
}
+ // Visible for testing
+ ColumnFamilyHandle offsetColumnFamilyHandle() {
+ return offsetColumnFamilyHandle;
+ }
+
private void wipeOffsets(final RocksDBStore.DBAccessor accessor) throws RocksDBException {
try (final RocksIterator iter = accessor.newIterator(offsetColumnFamilyHandle)) {
iter.seekToFirst();
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/DualColumnFamilyAccessor.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/DualColumnFamilyAccessor.java
index c7912aed0de..2ed7922b2e7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/DualColumnFamilyAccessor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/DualColumnFamilyAccessor.java
@@ -254,9 +254,25 @@ class DualColumnFamilyAccessor extends AbstractColumnFamilyAccessor {
@Override
public void close(final DBAccessor accessor) throws RocksDBException {
- super.close(accessor);
- oldColumnFamily.close();
- newColumnFamily.close();
+ try {
+ super.close(accessor);
+ } finally {
+ try {
+ oldColumnFamily.close();
+ } finally {
+ newColumnFamily.close();
+ }
+ }
+ }
+
+ // Visible for testing
+ ColumnFamilyHandle oldColumnFamily() {
+ return oldColumnFamily;
+ }
+
+ // Visible for testing
+ ColumnFamilyHandle newColumnFamily() {
+ return newColumnFamily;
}
private static class RocksDBDualCFIterator
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java
index c1f1e6ec150..ddaad20cdf0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeaders.java
@@ -61,7 +61,7 @@ public class RocksDBMigratingSessionStoreWithHeaders extends RocksDBStore implem
dbOptions,
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
new ColumnFamilyDescriptor(SESSION_STORE_HEADERS_VALUES_COLUMN_FAMILY_NAME, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
final ColumnFamilyHandle noHeadersColumnFamily = columnFamilies.get(0);
final ColumnFamilyHandle withHeadersColumnFamily = columnFamilies.get(1);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeaders.java
index e3dbf3d97b2..c3aabcae198 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeaders.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeaders.java
@@ -66,7 +66,7 @@ public class RocksDBMigratingWindowStoreWithHeaders extends RocksDBStore impleme
dbOptions,
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
new ColumnFamilyDescriptor(WINDOW_STORE_HEADERS_VALUES_COLUMN_FAMILY_NAME, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
final ColumnFamilyHandle noHeadersColumnFamily = columnFamilies.get(0);
final ColumnFamilyHandle withHeadersColumnFamily = columnFamilies.get(1);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index f207bd2139d..ede14506654 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -129,6 +129,7 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
private Cache cache;
private BloomFilter filter;
private Statistics statistics;
+ private ColumnFamilyOptions offsetsCfOptions;
private RocksDBConfigSetter configSetter;
private boolean userSpecifiedStatistics = false;
@@ -230,6 +231,12 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
fOptions = new FlushOptions();
fOptions.setWaitForFlush(true);
+ // The offsets CF stores only a small number of key-value pairs (one per changelog
+ // partition), so it does not need the heavyweight options used for the data CF;
+ offsetsCfOptions = new ColumnFamilyOptions();
+ offsetsCfOptions.setCompressionType(CompressionType.NO_COMPRESSION);
+ offsetsCfOptions.setWriteBufferSize(1024 * 1024L); // 1MB — sufficient for offset metadata
+
final Class<RocksDBConfigSetter> configSetterClass =
(Class<RocksDBConfigSetter>) configs.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG);
@@ -311,19 +318,14 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
}
/**
- * Creates lightweight {@link ColumnFamilyOptions} for the offsets column family. The offsets CF
- * stores only a small number of key-value pairs (one per changelog partition), so it does not
- * need the heavyweight options used for the data CF (large write buffers, bloom filters,
- * aggressive compaction). Sharing the data CF's options causes unnecessary write amplification
- * and compaction pressure that can contribute to RocksDB write stalls under heavy restore I/O.
+ * Returns the single read-only {@link ColumnFamilyOptions} instance used for the offsets CF.
+ *
+ * <p>The instance is created in {@link #openDB} before the open path runs
+ * {@code ColumnFamilyDescriptor} and is freed once in {@link #close()} /
+ * {@link #closeNativeResources()}.
*/
- protected static ColumnFamilyOptions createOffsetsCFOptions() {
- final ColumnFamilyOptions offsetsCFOptions = new ColumnFamilyOptions();
- offsetsCFOptions.setCompressionType(CompressionType.NO_COMPRESSION);
- offsetsCFOptions.setCompactionStyle(CompactionStyle.LEVEL);
- offsetsCFOptions.setWriteBufferSize(1024 * 1024L); // 1MB — sufficient for offset metadata
- offsetsCFOptions.setMaxWriteBufferNumber(2);
- return offsetsCFOptions;
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ return offsetsCfOptions;
}
void openRocksDB(final DBOptions dbOptions,
@@ -331,7 +333,7 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
final List<ColumnFamilyHandle> columnFamilies = openRocksDB(
dbOptions,
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
cfAccessor = new SingleColumnFamilyAccessor(columnFamilies.get(1), columnFamilies.get(0));
@@ -811,6 +813,9 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
dbAccessor.close();
db.close();
userSpecifiedOptions.close();
+ if (offsetsCfOptions != null) {
+ offsetsCfOptions.close();
+ }
wOptions.close();
fOptions.close();
filter.close();
@@ -822,6 +827,7 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
cfAccessor = null;
dbAccessor = null;
userSpecifiedOptions = null;
+ offsetsCfOptions = null;
wOptions = null;
fOptions = null;
db = null;
@@ -866,6 +872,10 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
userSpecifiedOptions.close();
userSpecifiedOptions = null;
}
+ if (offsetsCfOptions != null) {
+ offsetsCfOptions.close();
+ offsetsCfOptions = null;
+ }
if (wOptions != null) {
wOptions.close();
wOptions = null;
@@ -1156,8 +1166,16 @@ public class RocksDBStore implements KeyValueStore<Bytes, byte[]>, BatchWritingS
@Override
public void close(final RocksDBStore.DBAccessor accessor) throws RocksDBException {
- super.close(accessor);
- columnFamily.close();
+ try {
+ super.close(accessor);
+ } finally {
+ columnFamily.close();
+ }
+ }
+
+ // Visible for testing
+ ColumnFamilyHandle columnFamily() {
+ return columnFamily;
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
index 8771ec110c5..2b757e7a74c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java
@@ -57,7 +57,7 @@ public class RocksDBTimestampedStore extends RocksDBStore implements Timestamped
dbOptions,
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
new ColumnFamilyDescriptor(TIMESTAMPED_VALUES_COLUMN_FAMILY_NAME, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
final ColumnFamilyHandle noTimestampColumnFamily = columnFamilies.get(0);
final ColumnFamilyHandle withTimestampColumnFamily = columnFamilies.get(1);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
index e8f73af3fc1..9a53c35290a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
@@ -98,7 +98,7 @@ public class RocksDBTimestampedStoreWithHeaders extends RocksDBStore implements
dbOptions,
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
new ColumnFamilyDescriptor(TIMESTAMPED_VALUES_WITH_HEADERS_CF_NAME, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
final ColumnFamilyHandle defaultCf = columnFamilies.get(0);
@@ -139,7 +139,7 @@ public class RocksDBTimestampedStoreWithHeaders extends RocksDBStore implements
new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, columnFamilyOptions),
new ColumnFamilyDescriptor(LEGACY_TIMESTAMPED_CF_NAME, columnFamilyOptions),
new ColumnFamilyDescriptor(TIMESTAMPED_VALUES_WITH_HEADERS_CF_NAME, columnFamilyOptions),
- new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, createOffsetsCFOptions())
+ new ColumnFamilyDescriptor(OFFSETS_COLUMN_FAMILY_NAME, offsetsCFOptions())
);
try {
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java
index 48ef00a8190..3acc1aa06b0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java
@@ -47,6 +47,7 @@ import static java.util.Arrays.asList;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -536,4 +537,72 @@ public class RocksDBMigratingSessionStoreWithHeadersTest extends RocksDBStoreTes
"Expected header '" + header.key() + "' to match");
}
}
+
+ // KAFKA-20456 regression: the offsets-CF ColumnFamilyOptions must be released on close().
+ @Test
+ public void shouldCloseOffsetsCfOptionsOnStoreClose() {
+ final CapturingOffsetsMigratingSessionStore capturingStore = new CapturingOffsetsMigratingSessionStore();
+ rocksDBStore = capturingStore;
+ rocksDBStore.init(context, rocksDBStore);
+
+ final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions;
+ assertNotNull(captured, "offsetsCFOptions should have been invoked during init");
+ assertTrue(captured.isOwningHandle(),
+ "offsets CF options should own its native handle while store is open");
+
+ rocksDBStore.close();
+
+ assertFalse(captured.isOwningHandle(),
+ "offsets CF options native handle should be released by close()");
+ }
+
+ // KIP-1035 close-path leak regression: when the migrating store enters upgrade mode it
+ // installs a DualColumnFamilyAccessor with three CF handles (offsets + oldCF + newCF).
+ // If the close-time put(closedState) throws, all three handles must still be released
+ // via the try/finally chain in AbstractColumnFamilyAccessor.close() and
+ // DualColumnFamilyAccessor.close().
+ @Test
+ public void shouldCloseAllDualColumnFamilyHandlesWhenAccessorPutThrowsDuringClose() {
+ prepareDefaultStore();
+
+ rocksDBStore = new RocksDBMigratingSessionStoreWithHeaders(DB_NAME, METRICS_SCOPE);
+ rocksDBStore.init(context, rocksDBStore);
+
+ final DualColumnFamilyAccessor accessor = (DualColumnFamilyAccessor) rocksDBStore.cfAccessor;
+ final ColumnFamilyHandle offsetsHandle = accessor.offsetColumnFamilyHandle();
+ final ColumnFamilyHandle oldHandle = accessor.oldColumnFamily();
+ final ColumnFamilyHandle newHandle = accessor.newColumnFamily();
+ assertTrue(offsetsHandle.isOwningHandle());
+ assertTrue(oldHandle.isOwningHandle());
+ assertTrue(newHandle.isOwningHandle());
+
+ final ThrowingOnOffsetsPutDBAccessor wrapper =
+ new ThrowingOnOffsetsPutDBAccessor(rocksDBStore.dbAccessor, offsetsHandle);
+ rocksDBStore.dbAccessor = wrapper;
+
+ rocksDBStore.close();
+
+ assertTrue(wrapper.thrownPutCount.get() >= 1,
+ "expected the closedState put on the offsets CF to be invoked and to throw");
+ assertFalse(offsetsHandle.isOwningHandle(),
+ "offsets CF handle should still be closed when accessor.put throws");
+ assertFalse(oldHandle.isOwningHandle(),
+ "old CF handle should still be closed when super.close() throws");
+ assertFalse(newHandle.isOwningHandle(),
+ "new CF handle should still be closed when oldColumnFamily.close() chain runs");
+ }
+
+ private static final class CapturingOffsetsMigratingSessionStore extends RocksDBMigratingSessionStoreWithHeaders {
+ ColumnFamilyOptions capturedOffsetsOptions;
+
+ CapturingOffsetsMigratingSessionStore() {
+ super(DB_NAME, METRICS_SCOPE);
+ }
+
+ @Override
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ capturedOffsetsOptions = super.offsetsCFOptions();
+ return capturedOffsetsOptions;
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersCloseLeakTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersCloseLeakTest.java
new file mode 100644
index 00000000000..b44ab2f9e0a
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersCloseLeakTest.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecorder;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRocksDbConfigSetter;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+
+import java.io.File;
+import java.util.Properties;
+
+import static org.apache.kafka.streams.state.internals.RocksDBStore.DB_FILE_DIR;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Close-path leak regression tests for {@link RocksDBMigratingWindowStoreWithHeaders}, paralleling
+ * the migrating session-store tests:
+ *
+ * <ol>
+ * <li>KAFKA-20456 — the offsets-CF {@link ColumnFamilyOptions} must be released on
+ * {@code close()}.</li>
+ * <li>KIP-1035 close path — when the migrating window store opens against pre-existing
+ * default-CF data it installs a {@link DualColumnFamilyAccessor} with three CF handles
+ * (offsets + oldCF + newCF). If the close-time {@code put(closedState)} throws, all three
+ * handles must still be released via the try/finally chain.</li>
+ * </ol>
+ */
+public class RocksDBMigratingWindowStoreWithHeadersCloseLeakTest {
+ private static final String DB_NAME = "db-name";
+ private static final String METRICS_SCOPE = "metrics-scope";
+
+ private InternalMockProcessorContext<?, ?> context;
+ private RocksDBStore rocksDBStore;
+
+ @BeforeEach
+ public void setUp() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, MockRocksDbConfigSetter.class);
+ final File dir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ dir,
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (rocksDBStore != null) {
+ rocksDBStore.close();
+ }
+ }
+
+ @Test
+ public void shouldCloseOffsetsCfOptionsOnStoreClose() {
+ final CapturingOffsetsMigratingWindowStore capturingStore = new CapturingOffsetsMigratingWindowStore();
+ rocksDBStore = capturingStore;
+ rocksDBStore.init(context, rocksDBStore);
+
+ final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions;
+ assertNotNull(captured, "offsetsCFOptions should have been invoked during init");
+ assertTrue(captured.isOwningHandle(),
+ "offsets CF options should own its native handle while store is open");
+
+ rocksDBStore.close();
+
+ assertFalse(captured.isOwningHandle(),
+ "offsets CF options native handle should be released by close()");
+ }
+
+ @Test
+ public void shouldCloseAllDualColumnFamilyHandlesWhenAccessorPutThrowsDuringClose() {
+ // Seed the DEFAULT CF so the migrating store enters "upgrade mode" and installs a
+ // DualColumnFamilyAccessor rather than the single-CF accessor.
+ prepareLegacyDefaultStore();
+
+ rocksDBStore = new RocksDBMigratingWindowStoreWithHeaders(
+ DB_NAME, DB_FILE_DIR, new RocksDBMetricsRecorder(METRICS_SCOPE, DB_NAME));
+ rocksDBStore.init(context, rocksDBStore);
+
+ final DualColumnFamilyAccessor accessor = (DualColumnFamilyAccessor) rocksDBStore.cfAccessor;
+ final ColumnFamilyHandle offsetsHandle = accessor.offsetColumnFamilyHandle();
+ final ColumnFamilyHandle oldHandle = accessor.oldColumnFamily();
+ final ColumnFamilyHandle newHandle = accessor.newColumnFamily();
+ assertTrue(offsetsHandle.isOwningHandle());
+ assertTrue(oldHandle.isOwningHandle());
+ assertTrue(newHandle.isOwningHandle());
+
+ final ThrowingOnOffsetsPutDBAccessor wrapper =
+ new ThrowingOnOffsetsPutDBAccessor(rocksDBStore.dbAccessor, offsetsHandle);
+ rocksDBStore.dbAccessor = wrapper;
+
+ rocksDBStore.close();
+
+ assertTrue(wrapper.thrownPutCount.get() >= 1,
+ "expected the closedState put on the offsets CF to be invoked and to throw");
+ assertFalse(offsetsHandle.isOwningHandle(),
+ "offsets CF handle should still be closed when accessor.put throws");
+ assertFalse(oldHandle.isOwningHandle(),
+ "old CF handle should still be closed when super.close() throws");
+ assertFalse(newHandle.isOwningHandle(),
+ "new CF handle should still be closed when oldColumnFamily.close() chain runs");
+ }
+
+ private void prepareLegacyDefaultStore() {
+ final RocksDBStore kvStore = new RocksDBStore(DB_NAME, METRICS_SCOPE);
+ try {
+ kvStore.init(context, kvStore);
+ kvStore.put(new Bytes("seed-key".getBytes()), "seed-value".getBytes());
+ } finally {
+ kvStore.close();
+ }
+ }
+
+ private static final class CapturingOffsetsMigratingWindowStore extends RocksDBMigratingWindowStoreWithHeaders {
+ ColumnFamilyOptions capturedOffsetsOptions;
+
+ CapturingOffsetsMigratingWindowStore() {
+ super(DB_NAME, DB_FILE_DIR, new RocksDBMetricsRecorder(METRICS_SCOPE, DB_NAME));
+ }
+
+ @Override
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ capturedOffsetsOptions = super.offsetsCFOptions();
+ return capturedOffsetsOptions;
+ }
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreCloseLeakTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreCloseLeakTest.java
new file mode 100644
index 00000000000..2e3e2242a87
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreCloseLeakTest.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.MockRocksDbConfigSetter;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ColumnFamilyOptions;
+
+import java.io.File;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Regression tests for two RocksDBStore close-path native-memory leaks:
+ *
+ * <ol>
+ * <li>KAFKA-20456 — the {@link ColumnFamilyOptions} returned by
+ * {@code RocksDBStore#offsetsCFOptions()} must be released on {@code close()}.
+ * Constructing a {@code ColumnFamilyOptions} on the JNI side auto-allocates a default
+ * {@code BlockBasedTableFactory} and its {@code LRUCache}; the cache itself has no Java
+ * handle, so we assert on the options' own handle and rely on the destructor contract.</li>
+ * <li>KIP-1035 close-path — {@link AbstractColumnFamilyAccessor#close} writes a closed-state
+ * marker to the offsets CF. If that {@code put} throws (e.g. during an EOSv2 fencing
+ * cascade or unclean shutdown), the column family handles must still be released. We
+ * simulate the throw via {@link ThrowingOnOffsetsPutDBAccessor} and observe via
+ * {@code isOwningHandle()} because {@code RocksDBStore.close()} swallows the resulting
+ * {@code RocksDBException}.</li>
+ * </ol>
+ */
+public class RocksDBStoreCloseLeakTest {
+ private static final String DB_NAME = "db-name";
+ private static final String METRICS_SCOPE = "metrics-scope";
+
+ private InternalMockProcessorContext<?, ?> context;
+ private RocksDBStore rocksDBStore;
+
+ @BeforeEach
+ public void setUp() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, MockRocksDbConfigSetter.class);
+ final File dir = TestUtils.tempDirectory();
+ context = new InternalMockProcessorContext<>(
+ dir,
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (rocksDBStore != null) {
+ rocksDBStore.close();
+ }
+ }
+
+ @Test
+ public void shouldCloseOffsetsCfOptionsOnStoreClose() {
+ final CapturingOffsetsRocksDBStore capturingStore = new CapturingOffsetsRocksDBStore();
+ rocksDBStore = capturingStore;
+ rocksDBStore.init(context, rocksDBStore);
+
+ final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions;
+ assertNotNull(captured, "offsetsCFOptions should have been invoked during init");
+ assertTrue(captured.isOwningHandle(),
+ "offsets CF options should own its native handle while store is open");
+
+ rocksDBStore.close();
+
+ assertFalse(captured.isOwningHandle(),
+ "offsets CF options native handle should be released by close()");
+ }
+
+ @Test
+ public void shouldCloseColumnFamilyHandlesWhenAccessorPutThrowsDuringClose() {
+ rocksDBStore = new RocksDBStore(DB_NAME, METRICS_SCOPE);
+ rocksDBStore.init(context, rocksDBStore);
+
+ final RocksDBStore.SingleColumnFamilyAccessor accessor =
+ (RocksDBStore.SingleColumnFamilyAccessor) rocksDBStore.cfAccessor;
+ final ColumnFamilyHandle offsetsHandle = accessor.offsetColumnFamilyHandle();
+ final ColumnFamilyHandle dataHandle = accessor.columnFamily();
+ assertTrue(offsetsHandle.isOwningHandle());
+ assertTrue(dataHandle.isOwningHandle());
+
+ final ThrowingOnOffsetsPutDBAccessor wrapper =
+ new ThrowingOnOffsetsPutDBAccessor(rocksDBStore.dbAccessor, offsetsHandle);
+ rocksDBStore.dbAccessor = wrapper;
+
+ rocksDBStore.close();
+
+ assertTrue(wrapper.thrownPutCount.get() >= 1,
+ "expected the closedState put on the offsets CF to be invoked and to throw");
+ assertFalse(offsetsHandle.isOwningHandle(),
+ "offsets CF handle should still be closed when accessor.put throws");
+ assertFalse(dataHandle.isOwningHandle(),
+ "data CF handle should still be closed when super.close() throws");
+ }
+
+ private static final class CapturingOffsetsRocksDBStore extends RocksDBStore {
+ ColumnFamilyOptions capturedOffsetsOptions;
+
+ CapturingOffsetsRocksDBStore() {
+ super(DB_NAME, METRICS_SCOPE);
+ }
+
+ @Override
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ capturedOffsetsOptions = super.offsetsCFOptions();
+ return capturedOffsetsOptions;
+ }
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreTest.java
index 20cc89d45af..3c8f87bda49 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreTest.java
@@ -44,7 +44,9 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class RocksDBTimestampedStoreTest extends RocksDBStoreTest {
@@ -515,4 +517,38 @@ public class RocksDBTimestampedStoreTest extends RocksDBStoreTest {
keyValueStore.close();
}
}
+
+ // KAFKA-20456 regression: the offsets-CF ColumnFamilyOptions allocated by
+ // offsetsCFOptions() must be released on close(); otherwise the JNI-auto-allocated
+ // default BlockBasedTableFactory and its LRUCache leak per store open.
+ @Test
+ public void shouldCloseOffsetsCfOptionsOnStoreClose() {
+ final CapturingOffsetsTimestampedStore capturingStore = new CapturingOffsetsTimestampedStore();
+ rocksDBStore = capturingStore;
+ rocksDBStore.init(context, rocksDBStore);
+
+ final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions;
+ assertNotNull(captured, "offsetsCFOptions should have been invoked during init");
+ assertTrue(captured.isOwningHandle(),
+ "offsets CF options should own its native handle while store is open");
+
+ rocksDBStore.close();
+
+ assertFalse(captured.isOwningHandle(),
+ "offsets CF options native handle should be released by close()");
+ }
+
+ private static final class CapturingOffsetsTimestampedStore extends RocksDBTimestampedStore {
+ ColumnFamilyOptions capturedOffsetsOptions;
+
+ CapturingOffsetsTimestampedStore() {
+ super(DB_NAME, METRICS_SCOPE);
+ }
+
+ @Override
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ capturedOffsetsOptions = super.offsetsCFOptions();
+ return capturedOffsetsOptions;
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
index ef491c34fe6..d098696ded7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
@@ -1046,4 +1046,50 @@ public class RocksDBTimestampedStoreWithHeadersTest extends RocksDBStoreTest {
// Verify no execution info was collected
assertTrue(result.getExecutionInfo().isEmpty(), "Expected no execution info to be collected");
}
+
+ // KAFKA-20456 regression: both RocksDBTimestampedStoreWithHeaders open paths
+ // (openFromDefaultStore and openFromTimestampedStore) independently call
+ // offsetsCFOptions(); the returned ColumnFamilyOptions must be released on close().
+
+ @Test
+ public void shouldCloseOffsetsCfOptionsAfterOpenFromDefaultStore() {
+ prepareKeyValueStoreWithMultipleKeys();
+ assertOffsetsCfOptionsClosedAfterStoreClose();
+ }
+
+ @Test
+ public void shouldCloseOffsetsCfOptionsAfterOpenFromTimestampedStore() {
+ prepareTimestampedStore();
+ assertOffsetsCfOptionsClosedAfterStoreClose();
+ }
+
+ private void assertOffsetsCfOptionsClosedAfterStoreClose() {
+ final CapturingOffsetsHeadersStore capturingStore = new CapturingOffsetsHeadersStore();
+ rocksDBStore = capturingStore;
+ rocksDBStore.init(context, rocksDBStore);
+
+ final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions;
+ assertNotNull(captured, "offsetsCFOptions should have been invoked during init");
+ assertTrue(captured.isOwningHandle(),
+ "offsets CF options should own its native handle while store is open");
+
+ rocksDBStore.close();
+
+ assertFalse(captured.isOwningHandle(),
+ "offsets CF options native handle should be released by close()");
+ }
+
+ private static final class CapturingOffsetsHeadersStore extends RocksDBTimestampedStoreWithHeaders {
+ ColumnFamilyOptions capturedOffsetsOptions;
+
+ CapturingOffsetsHeadersStore() {
+ super(DB_NAME, METRICS_SCOPE);
+ }
+
+ @Override
+ protected ColumnFamilyOptions offsetsCFOptions() {
+ capturedOffsetsOptions = super.offsetsCFOptions();
+ return capturedOffsetsOptions;
+ }
+ }
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/ThrowingOnOffsetsPutDBAccessor.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThrowingOnOffsetsPutDBAccessor.java
new file mode 100644
index 00000000000..ae0525105c8
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/ThrowingOnOffsetsPutDBAccessor.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.ReadOptions;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Delegating {@link RocksDBStore.DBAccessor} that throws {@link RocksDBException} on any
+ * {@code put} against the offsets column family. Used to simulate the close-time
+ * {@code put(closedState)} failure observed in production (e.g. EOSv2 fencing cascades or
+ * unclean shutdowns) and verify that column family handles are still released on the
+ * exception path via the try/finally ordering in {@code AbstractColumnFamilyAccessor.close()},
+ * {@code SingleColumnFamilyAccessor.close()}, and {@code DualColumnFamilyAccessor.close()}.
+ */
+final class ThrowingOnOffsetsPutDBAccessor implements RocksDBStore.DBAccessor {
+ private final RocksDBStore.DBAccessor delegate;
+ private final ColumnFamilyHandle offsetsHandle;
+ final AtomicInteger thrownPutCount = new AtomicInteger();
+
+ ThrowingOnOffsetsPutDBAccessor(final RocksDBStore.DBAccessor delegate,
+ final ColumnFamilyHandle offsetsHandle) {
+ this.delegate = delegate;
+ this.offsetsHandle = offsetsHandle;
+ }
+
+ @Override
+ public byte[] get(final ColumnFamilyHandle columnFamily, final byte[] key) throws RocksDBException {
+ return delegate.get(columnFamily, key);
+ }
+
+ @Override
+ public byte[] get(final ColumnFamilyHandle columnFamily, final ReadOptions readOptions, final byte[] key) throws RocksDBException {
+ return delegate.get(columnFamily, readOptions, key);
+ }
+
+ @Override
+ public RocksIterator newIterator(final ColumnFamilyHandle columnFamily) {
+ return delegate.newIterator(columnFamily);
+ }
+
+ @Override
+ public void put(final ColumnFamilyHandle columnFamily, final byte[] key, final byte[] value) throws RocksDBException {
+ if (columnFamily == offsetsHandle) {
+ thrownPutCount.incrementAndGet();
+ throw new RocksDBException("simulated put failure on offsets CF");
+ }
+ delegate.put(columnFamily, key, value);
+ }
+
+ @Override
+ public void delete(final ColumnFamilyHandle columnFamily, final byte[] key) throws RocksDBException {
+ delegate.delete(columnFamily, key);
+ }
+
+ @Override
+ public void deleteRange(final ColumnFamilyHandle columnFamily, final byte[] from, final byte[] to) throws RocksDBException {
+ delegate.deleteRange(columnFamily, from, to);
+ }
+
+ @Override
+ public long approximateNumEntries(final ColumnFamilyHandle columnFamily) throws RocksDBException {
+ return delegate.approximateNumEntries(columnFamily);
+ }
+
+ @Override
+ public void flush(final ColumnFamilyHandle... columnFamilies) throws RocksDBException {
+ delegate.flush(columnFamilies);
+ }
+
+ @Override
+ public void reset() {
+ delegate.reset();
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+}