From 5b55f7465bb84abd80c6ae7e227a26f5f7381467 Mon Sep 17 00:00:00 2001 From: Bill Date: Tue, 26 May 2026 16:05:41 -0400 Subject: [PATCH 1/4] Close offsets-CF ColumnFamilyOptions and ensure CF handles are released on close-path exceptions to fix native memory leak under EOSv2 cascades --- .../AbstractColumnFamilyAccessor.java | 18 +- .../internals/DualColumnFamilyAccessor.java | 22 ++- .../streams/state/internals/RocksDBStore.java | 47 +++++- ...BMigratingSessionStoreWithHeadersTest.java | 69 ++++++++ ...ngWindowStoreWithHeadersCloseLeakTest.java | 156 ++++++++++++++++++ .../internals/RocksDBStoreCloseLeakTest.java | 139 ++++++++++++++++ .../RocksDBTimestampedStoreTest.java | 36 ++++ ...ocksDBTimestampedStoreWithHeadersTest.java | 46 ++++++ .../ThrowingOnOffsetsPutDBAccessor.java | 98 +++++++++++ 9 files changed, 615 insertions(+), 16 deletions(-) create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingWindowStoreWithHeadersCloseLeakTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreCloseLeakTest.java create mode 100644 streams/src/test/java/org/apache/kafka/streams/state/internals/ThrowingOnOffsetsPutDBAccessor.java 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 2cf776423a0d0..eb9958b3a7ff0 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 @@ -99,11 +99,16 @@ public final Position open(final RocksDBStore.DBAccessor accessor, final boolean 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 @@ -116,6 +121,11 @@ public final Long getCommittedOffset(final RocksDBStore.DBAccessor accessor, fin } + // 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 cb23ef9134b6b..10664c1326b27 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 @@ -246,9 +246,25 @@ public void addToBatch(final byte[] key, @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/RocksDBStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java index 7ecc9d7f09906..f4f092c524945 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,10 @@ public class RocksDBStore implements KeyValueStore, BatchWritingS private Cache cache; private BloomFilter filter; private Statistics statistics; + // Options for the offsets column family, kept on a field so close() can release the + // native resources (notably the BlockBasedTableFactory's default LRUCache) that the + // RocksDB C++ side auto-allocates when a ColumnFamilyOptions is constructed. + private ColumnFamilyOptions offsetsCfOptions; private RocksDBConfigSetter configSetter; private boolean userSpecifiedStatistics = false; @@ -316,14 +320,23 @@ private void addValueProvidersToMetricsRecorder() { * 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. + * + *

The returned {@code ColumnFamilyOptions} is stored on {@link #offsetsCfOptions} so that + * {@link #close()} can release it. Constructing a {@code ColumnFamilyOptions} on the JNI side + * auto-allocates a default {@code BlockBasedTableFactory} and its {@code LRUCache}; if the + * options object is never closed, that native cache leaks. We also explicitly disable the block + * cache for this CF because the offsets metadata is tiny and benefits nothing from caching. */ - 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 createOffsetsCFOptions() { + offsetsCfOptions = new ColumnFamilyOptions(); + offsetsCfOptions.setCompressionType(CompressionType.NO_COMPRESSION); + offsetsCfOptions.setCompactionStyle(CompactionStyle.LEVEL); + offsetsCfOptions.setWriteBufferSize(1024 * 1024L); // 1MB — sufficient for offset metadata + offsetsCfOptions.setMaxWriteBufferNumber(2); + final BlockBasedTableConfig offsetsTableConfig = new BlockBasedTableConfig(); + offsetsTableConfig.setNoBlockCache(true); + offsetsCfOptions.setTableFormatConfig(offsetsTableConfig); + return offsetsCfOptions; } void openRocksDB(final DBOptions dbOptions, @@ -811,6 +824,9 @@ public synchronized void close() { dbAccessor.close(); db.close(); userSpecifiedOptions.close(); + if (offsetsCfOptions != null) { + offsetsCfOptions.close(); + } wOptions.close(); fOptions.close(); filter.close(); @@ -822,6 +838,7 @@ public synchronized void close() { cfAccessor = null; dbAccessor = null; userSpecifiedOptions = null; + offsetsCfOptions = null; wOptions = null; fOptions = null; db = null; @@ -866,6 +883,10 @@ private void closeOptionsAndFilters() { userSpecifiedOptions.close(); userSpecifiedOptions = null; } + if (offsetsCfOptions != null) { + offsetsCfOptions.close(); + offsetsCfOptions = null; + } if (wOptions != null) { wOptions.close(); wOptions = null; @@ -1165,8 +1186,16 @@ public void addToBatch(final byte[] key, @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/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBMigratingSessionStoreWithHeadersTest.java index 48ef00a819058..1d36629ac87fc 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 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 @@ private void assertValueWithHeaders(final byte[] value, final String expectedAgg "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, "createOffsetsCFOptions 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 createOffsetsCFOptions() { + capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 0000000000000..724013c2aae12 --- /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: + * + *

    + *
  1. KAFKA-20456 — the offsets-CF {@link ColumnFamilyOptions} must be released on + * {@code close()}.
    2. + *
  2. 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.
    3. + *
+ */ +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, "createOffsetsCFOptions 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 createOffsetsCFOptions() { + capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 0000000000000..9e4b82018f00a --- /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: + * + *
    + *
  1. KAFKA-20456 — the {@link ColumnFamilyOptions} returned by + * {@code RocksDBStore#createOffsetsCFOptions()} 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.
    2. + *
  2. 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}.
    3. + *
+ */ +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, "createOffsetsCFOptions 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 createOffsetsCFOptions() { + capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 20cc89d45afaa..385127046aac1 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.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 @@ private void prepareOldStore() { keyValueStore.close(); } } + + // KAFKA-20456 regression: the offsets-CF ColumnFamilyOptions allocated by + // createOffsetsCFOptions() 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, "createOffsetsCFOptions 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 createOffsetsCFOptions() { + capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 ef491c34fe60e..dcea231cae0d9 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 void shouldNotCollectExecutionInfoWhenNotRequested() { // 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 + // createOffsetsCFOptions(); 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, "createOffsetsCFOptions 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 createOffsetsCFOptions() { + capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 0000000000000..ae0525105c87c --- /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(); + } +} From c007845d7eb781b832c8a155905da6e3785ff8c7 Mon Sep 17 00:00:00 2001 From: Bill Date: Wed, 27 May 2026 11:04:36 -0400 Subject: [PATCH 2/4] Address comments, simplify configs --- .../streams/state/internals/RocksDBStore.java | 23 ++++--------------- 1 file changed, 4 insertions(+), 19 deletions(-) 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 f4f092c524945..934f8ac3601d9 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,9 +129,6 @@ public class RocksDBStore implements KeyValueStore, BatchWritingS private Cache cache; private BloomFilter filter; private Statistics statistics; - // Options for the offsets column family, kept on a field so close() can release the - // native resources (notably the BlockBasedTableFactory's default LRUCache) that the - // RocksDB C++ side auto-allocates when a ColumnFamilyOptions is constructed. private ColumnFamilyOptions offsetsCfOptions; private RocksDBConfigSetter configSetter; @@ -315,27 +312,15 @@ private void addValueProvidersToMetricsRecorder() { } /** - * 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. - * - *

The returned {@code ColumnFamilyOptions} is stored on {@link #offsetsCfOptions} so that - * {@link #close()} can release it. Constructing a {@code ColumnFamilyOptions} on the JNI side - * auto-allocates a default {@code BlockBasedTableFactory} and its {@code LRUCache}; if the - * options object is never closed, that native cache leaks. We also explicitly disable the block - * cache for this CF because the offsets metadata is tiny and benefits nothing from caching. + * 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. + * Uses the default options for compaction and max write buffer number. */ protected ColumnFamilyOptions createOffsetsCFOptions() { offsetsCfOptions = new ColumnFamilyOptions(); offsetsCfOptions.setCompressionType(CompressionType.NO_COMPRESSION); - offsetsCfOptions.setCompactionStyle(CompactionStyle.LEVEL); offsetsCfOptions.setWriteBufferSize(1024 * 1024L); // 1MB — sufficient for offset metadata - offsetsCfOptions.setMaxWriteBufferNumber(2); - final BlockBasedTableConfig offsetsTableConfig = new BlockBasedTableConfig(); - offsetsTableConfig.setNoBlockCache(true); - offsetsCfOptions.setTableFormatConfig(offsetsTableConfig); return offsetsCfOptions; } From eed4fb6d10b68d3139a14e76b3196f1f99af02e1 Mon Sep 17 00:00:00 2001 From: Bill Date: Fri, 5 Jun 2026 15:16:03 -0400 Subject: [PATCH 3/4] Make offsetsCFOptions singleton shared via accessor, safer approach --- ...cksDBMigratingSessionStoreWithHeaders.java | 2 +- ...ocksDBMigratingWindowStoreWithHeaders.java | 2 +- .../streams/state/internals/RocksDBStore.java | 23 +++++++++++-------- .../internals/RocksDBTimestampedStore.java | 2 +- .../RocksDBTimestampedStoreWithHeaders.java | 4 ++-- ...BMigratingSessionStoreWithHeadersTest.java | 6 ++--- ...ngWindowStoreWithHeadersCloseLeakTest.java | 6 ++--- .../internals/RocksDBStoreCloseLeakTest.java | 8 +++---- .../RocksDBTimestampedStoreTest.java | 8 +++---- ...ocksDBTimestampedStoreWithHeadersTest.java | 8 +++---- 10 files changed, 37 insertions(+), 32 deletions(-) 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 c1f1e6ec150c1..ddaad20cdf068 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 @@ void openRocksDB(final DBOptions dbOptions, 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 8789272ba3745..5402c377d5773 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 @@ -68,7 +68,7 @@ void openRocksDB(final DBOptions dbOptions, 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 934f8ac3601d9..f178684ee99ac 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 @@ -231,6 +231,12 @@ void openDB(final Map configs, final File stateDir) { 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 configSetterClass = (Class) configs.get(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG); @@ -312,15 +318,14 @@ private void addValueProvidersToMetricsRecorder() { } /** - * 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. - * Uses the default options for compaction and max write buffer number. + * Returns the single read-only {@link ColumnFamilyOptions} instance used for the offsets CF. + * + *

The instance is created in {@link #openDB} before the open path runs, so this is a pure + * getter (no lazy initialization, hence no data race). The same instance backs every offsets + * {@code ColumnFamilyDescriptor} and is freed once in {@link #close()} / + * {@link #closeNativeResources()}. */ - protected ColumnFamilyOptions createOffsetsCFOptions() { - offsetsCfOptions = new ColumnFamilyOptions(); - offsetsCfOptions.setCompressionType(CompressionType.NO_COMPRESSION); - offsetsCfOptions.setWriteBufferSize(1024 * 1024L); // 1MB — sufficient for offset metadata + protected ColumnFamilyOptions offsetsCFOptions() { return offsetsCfOptions; } @@ -329,7 +334,7 @@ void openRocksDB(final DBOptions dbOptions, final List 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)); 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 8771ec110c57d..2b757e7a74c6f 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 @@ void openRocksDB(final DBOptions dbOptions, 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 e8f73af3fc1f8..9a53c35290a3e 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 @@ private void openFromDefaultStore(final DBOptions dbOptions, 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 @@ private void openFromTimestampedStore(final DBOptions dbOptions, 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 1d36629ac87fc..3acc1aa06b0a1 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 @@ -546,7 +546,7 @@ public void shouldCloseOffsetsCfOptionsOnStoreClose() { rocksDBStore.init(context, rocksDBStore); final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions; - assertNotNull(captured, "createOffsetsCFOptions should have been invoked during init"); + assertNotNull(captured, "offsetsCFOptions should have been invoked during init"); assertTrue(captured.isOwningHandle(), "offsets CF options should own its native handle while store is open"); @@ -600,8 +600,8 @@ private static final class CapturingOffsetsMigratingSessionStore extends RocksDB } @Override - protected ColumnFamilyOptions createOffsetsCFOptions() { - capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 index 724013c2aae12..b44ab2f9e0a37 100644 --- 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 @@ -86,7 +86,7 @@ public void shouldCloseOffsetsCfOptionsOnStoreClose() { rocksDBStore.init(context, rocksDBStore); final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions; - assertNotNull(captured, "createOffsetsCFOptions should have been invoked during init"); + assertNotNull(captured, "offsetsCFOptions should have been invoked during init"); assertTrue(captured.isOwningHandle(), "offsets CF options should own its native handle while store is open"); @@ -148,8 +148,8 @@ private static final class CapturingOffsetsMigratingWindowStore extends RocksDBM } @Override - protected ColumnFamilyOptions createOffsetsCFOptions() { - capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 index 9e4b82018f00a..2e3e2242a8780 100644 --- 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 @@ -41,7 +41,7 @@ * *

    *
  1. KAFKA-20456 — the {@link ColumnFamilyOptions} returned by - * {@code RocksDBStore#createOffsetsCFOptions()} must be released on {@code close()}. + * {@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.
    2. @@ -87,7 +87,7 @@ public void shouldCloseOffsetsCfOptionsOnStoreClose() { rocksDBStore.init(context, rocksDBStore); final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions; - assertNotNull(captured, "createOffsetsCFOptions should have been invoked during init"); + assertNotNull(captured, "offsetsCFOptions should have been invoked during init"); assertTrue(captured.isOwningHandle(), "offsets CF options should own its native handle while store is open"); @@ -131,8 +131,8 @@ private static final class CapturingOffsetsRocksDBStore extends RocksDBStore { } @Override - protected ColumnFamilyOptions createOffsetsCFOptions() { - capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 385127046aac1..3c8f87bda498e 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 @@ -519,7 +519,7 @@ private void prepareOldStore() { } // KAFKA-20456 regression: the offsets-CF ColumnFamilyOptions allocated by - // createOffsetsCFOptions() must be released on close(); otherwise the JNI-auto-allocated + // offsetsCFOptions() must be released on close(); otherwise the JNI-auto-allocated // default BlockBasedTableFactory and its LRUCache leak per store open. @Test public void shouldCloseOffsetsCfOptionsOnStoreClose() { @@ -528,7 +528,7 @@ public void shouldCloseOffsetsCfOptionsOnStoreClose() { rocksDBStore.init(context, rocksDBStore); final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions; - assertNotNull(captured, "createOffsetsCFOptions should have been invoked during init"); + assertNotNull(captured, "offsetsCFOptions should have been invoked during init"); assertTrue(captured.isOwningHandle(), "offsets CF options should own its native handle while store is open"); @@ -546,8 +546,8 @@ private static final class CapturingOffsetsTimestampedStore extends RocksDBTimes } @Override - protected ColumnFamilyOptions createOffsetsCFOptions() { - capturedOffsetsOptions = super.createOffsetsCFOptions(); + 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 dcea231cae0d9..d098696ded7e9 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 @@ -1049,7 +1049,7 @@ public void shouldNotCollectExecutionInfoWhenNotRequested() { // KAFKA-20456 regression: both RocksDBTimestampedStoreWithHeaders open paths // (openFromDefaultStore and openFromTimestampedStore) independently call - // createOffsetsCFOptions(); the returned ColumnFamilyOptions must be released on close(). + // offsetsCFOptions(); the returned ColumnFamilyOptions must be released on close(). @Test public void shouldCloseOffsetsCfOptionsAfterOpenFromDefaultStore() { @@ -1069,7 +1069,7 @@ private void assertOffsetsCfOptionsClosedAfterStoreClose() { rocksDBStore.init(context, rocksDBStore); final ColumnFamilyOptions captured = capturingStore.capturedOffsetsOptions; - assertNotNull(captured, "createOffsetsCFOptions should have been invoked during init"); + assertNotNull(captured, "offsetsCFOptions should have been invoked during init"); assertTrue(captured.isOwningHandle(), "offsets CF options should own its native handle while store is open"); @@ -1087,8 +1087,8 @@ private static final class CapturingOffsetsHeadersStore extends RocksDBTimestamp } @Override - protected ColumnFamilyOptions createOffsetsCFOptions() { - capturedOffsetsOptions = super.createOffsetsCFOptions(); + protected ColumnFamilyOptions offsetsCFOptions() { + capturedOffsetsOptions = super.offsetsCFOptions(); return capturedOffsetsOptions; } } From 9a6cadd17f91188d81228c5e32f0bec0198eb2fb Mon Sep 17 00:00:00 2001 From: Bill Date: Fri, 5 Jun 2026 15:40:13 -0400 Subject: [PATCH 4/4] Clean up comments --- .../org/apache/kafka/streams/state/internals/RocksDBStore.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 f178684ee99ac..28b8976b18717 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 @@ -320,8 +320,7 @@ private void addValueProvidersToMetricsRecorder() { /** * Returns the single read-only {@link ColumnFamilyOptions} instance used for the offsets CF. * - *

    The instance is created in {@link #openDB} before the open path runs, so this is a pure - * getter (no lazy initialization, hence no data race). The same instance backs every offsets + *

    The instance is created in {@link #openDB} before the open path runs * {@code ColumnFamilyDescriptor} and is freed once in {@link #close()} / * {@link #closeNativeResources()}. */