KAFKA-20132: Implement TimestampedKeyValueStoreWithHeaders (2/N) by aliehsaeedii · Pull Request #21446 · apache/kafka

aliehsaeedii · 2026-02-10T15:00:38Z

This PR add RocksDBTimestampedStoreWithHeaders and the corresponding
unit test (RocksDBTimestampedStoreWithHeadersTest) for the
TimestampedKeyValueStoreWithHeaders introduced in KIP-1271.

Reviewers: TengYao Chi frankvicky@apache.org, Matthias J. Sax
matthias@confluent.io

frankvicky

Thanks for the PR.

frankvicky · 2026-02-10T21:46:59Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+        // Check if we're upgrading from RocksDBTimestampedStore (which uses keyValueWithTimestamp CF)
+        final List<byte[]> existingCFs;
+        try {
+            final org.rocksdb.Options options = new org.rocksdb.Options(dbOptions, new ColumnFamilyOptions());


Could we replace full qualified name with import?

frankvicky · 2026-02-10T21:48:27Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+            existingCFs = RocksDB.listColumnFamilies(options, dbDir.getAbsolutePath());
+            options.close();
+        } catch (final org.rocksdb.RocksDBException e) {
+            throw new org.apache.kafka.streams.errors.ProcessorStateException("Error listing column families for store " + name, e);


frankvicky · 2026-02-10T21:48:34Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+        }
+
+        final boolean upgradingFromTimestampedStore = existingCFs.stream()
+            .anyMatch(cf -> java.util.Arrays.equals(cf, LEGACY_TIMESTAMPED_CF_NAME));


frankvicky · 2026-02-10T21:54:31Z

...st/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java

+        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(RocksDBTimestampedStoreWithHeaders.class)) {
+            rocksDBStore.init(context, rocksDBStore);
+
+            assertThat(appender.getMessages(), hasItem("Opening store " + DB_NAME + " in regular headers-aware mode"));


I'm wondering should we keep using hamcrest in test.
AFAIK, hamcrest is not active, maybe we should use JUnit instead?
cc. @mjsax @bbejeck

I never pay close attention to this, and I am personally fine either way. Of course, if there is some overall effort to move off hamcrest in favor of JUnit, it would make sense to avoid adding hamcrest code.

frankvicky

Thanks for the PR.
There are lots of descriptive comments in the test.
Could you please consider to put them in the assertion message?

For example:

        // approx: 7 entries on legacy timestamped CF, 0 in new headers-aware CF
        assertEquals(7L, rocksDBStore.approximateNumEntries());

to

        assertEquals(7L, rocksDBStore.approximateNumEntries(), "should have 7 entries on legacy timestamped");

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

frankvicky · 2026-02-11T12:35:30Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+    private static final byte[] LEGACY_TIMESTAMPED_CF_NAME =
+        "keyValueWithTimestamp".getBytes(StandardCharsets.UTF_8);
+
+    // New column family for header-aware timestamped values.


Suggested change

// New column family for header-aware timestamped values.

/**

* New column family for header-aware timestamped values.

*/

This comment seems to not add any value -- I would remove it.

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

mjsax · 2026-02-11T22:22:53Z

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

+    private static final byte[] LEGACY_TIMESTAMPED_CF_NAME =
+        "keyValueWithTimestamp".getBytes(StandardCharsets.UTF_8);
+
+    // New column family for header-aware timestamped values.


This comment seems to not add any value -- I would remove it.

...c/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java

mjsax · 2026-02-11T23:44:53Z

...st/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java

+            {
+                final KeyValue<Bytes, byte[]> keyValue = itAll.next();
+                assertArrayEquals("key4".getBytes(), keyValue.key.get());
+                assertEquals(13, keyValue.value.length, "Expected header-aware format: varint(0) + empty headers(0) + timestamp(8) + value(4) = 13 bytes for key4 from legacy CF");


Similar to above. putIfAbsent would not update the value, because the key exists, but still migrate the value from legacy- to headers-CF?

iterator does the conversion on the fly but does not apply it to the physical store.

mjsax · 2026-02-11T23:45:26Z

...st/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java

+            {
+                final KeyValue<Bytes, byte[]> keyValue = itAll.next();
+                assertArrayEquals("key5".getBytes(), keyValue.key.get());
+                assertEquals(14, keyValue.value.length, "Expected header-aware format: varint(0) + empty headers(0) + timestamp(8) + value(5) = 14 bytes for key5 from legacy CF");


Same as for key4

...st/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java

mjsax · 2026-02-12T00:11:59Z

Please rebase to trunk to ensure we don't miss anything about the checkstyle/spotbugs fix.

…/RocksDBTimestampedStoreWithHeaders.java Co-authored-by: TengYao Chi <kitingiao@gmail.com>

add headers-aware rocksdb

b802101

github-actions bot added triage PRs from the community streams labels Feb 10, 2026

mjsax added kip Requires or implements a KIP ci-approved and removed triage PRs from the community labels Feb 10, 2026

frankvicky reviewed Feb 10, 2026

View reviewed changes

aliehsaeedii added 4 commits February 11, 2026 09:22

use import instead of qualified name

3703df0

Merge branch 'trunk' into hrocksdb

62ff59a

Merge branch 'trunk' into hrocksdb

1e8dcfc

replace junit

f52bdf6

frankvicky reviewed Feb 11, 2026

View reviewed changes

address reviews

0a34325

aliehsaeedii mentioned this pull request Feb 11, 2026

KAFKA-20132: Implement TimestampedKeyValueStoreWithHeaders (5/N) #21455

Open

mjsax reviewed Feb 12, 2026

View reviewed changes

aliehsaeedii and others added 3 commits February 12, 2026 12:23

Update streams/src/main/java/org/apache/kafka/streams/state/internals…

daba6fc

…/RocksDBTimestampedStoreWithHeaders.java Co-authored-by: TengYao Chi <kitingiao@gmail.com>

Update streams/src/main/java/org/apache/kafka/streams/state/internals…

b5ce4e7

…/RocksDBTimestampedStoreWithHeaders.java Co-authored-by: TengYao Chi <kitingiao@gmail.com>

Update streams/src/main/java/org/apache/kafka/streams/state/internals…

99642ec

…/RocksDBTimestampedStoreWithHeaders.java Co-authored-by: TengYao Chi <kitingiao@gmail.com>

frankvicky mentioned this pull request Feb 12, 2026

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (1/N) #21465

Open

aliehsaeedii and others added 6 commits February 13, 2026 12:17

address reviews

dc3c463

Delete streams/hs_err_pid19686.log

1043503

Remove accidentally committed hs_err crash logs

549f498

say why 22

427a66f

Merge branch 'trunk' into hrocksdb

8b840f2

fix cs errors

97e6fa8

Conversation

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

Uh oh!

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii Feb 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mjsax commented Feb 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

aliehsaeedii commented Feb 10, 2026 •

edited by github-actions bot

Loading

aliehsaeedii Feb 13, 2026 •

edited

Loading