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

aliehsaeedii · 2026-02-11T17:15:11Z

This PR adds required classes or modifies the existing ones to build
the TimestampedKeyValueStoreWithHeaders introduced in
KIP-1271.
It include utests as well.

github-actions · 2026-02-19T03:53:04Z

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

frankvicky

Thanks for the PR.

frankvicky · 2026-02-19T13:58:07Z

...rc/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java

+        return returnHeadersStore ?
+            new RocksDBTimestampedStoreWithHeaders(name, metricsScope()) :
+            returnTimestampedStore ?
            new RocksDBTimestampedStore(name, metricsScope()) :
            new RocksDBStore(name, metricsScope());


Given this condition is getting complex, should we use if statement?
For example

if (returnHeadersStore) return new RocksDBTimestampedStoreWithHeaders(name, metricsScope()); if (returnTimestampedStore) return new RocksDBTimestampedStore(name, metricsScope()); return new RocksDBStore(name, metricsScope());

frankvicky · 2026-02-19T14:01:26Z

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

+ * {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders} and
+ * {@link org.apache.kafka.streams.state.TimestampedKeyValueStore}.


Could you please use import instead of full qualified name?

frankvicky · 2026-02-19T14:04:37Z

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

+ * {@link org.apache.kafka.streams.state.TimestampedKeyValueStore}.
+ * <p>
+ * If a user provides a supplier for {@code TimestampedKeyValueStore} (without headers) via
+ * {@link org.apache.kafka.streams.kstream.Materialized#as(KeyValueBytesStoreSupplier)} when building


frankvicky · 2026-02-19T14:12:52Z

...org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java

+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;


As previous PRs, we should avoid using hamcrest

frankvicky · 2026-02-19T14:35:05Z

.../test/java/org/apache/kafka/streams/integration/TimestampedKeyValueStoreWithHeadersTest.java

+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;


As previous PRs, we should avoid using hamcrest

streams/src/main/java/org/apache/kafka/streams/state/Stores.java

mjsax

Made a pass (skipped tests for now)

mjsax · 2026-02-19T22:52:12Z

...rc/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java

+                                             final boolean returnHeadersStore) {
        this.name = name;
        this.returnTimestampedStore = returnTimestampedStore;
+        this.returnHeadersStore = returnHeadersStore;


Should we add a check for invalid combination of returnTimestampedStore and returnHeadersStore?

We only support 3 out of 4 combinations, right?

Due to the first else, the 4th combination (both true) is served with the first if which sematically is not wrong as headers stores are both ts and are WithHeaders. No idea if considering 4th combination helps. Should we throw exception for now?

Should we throw exception for now?

That was the idea.

mjsax · 2026-02-19T22:58:52Z

...ava/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeaders.java

+        if (!enableCaching) {
+            return inner;
+        }
+        return new CachingKeyValueStore(inner, true);


I was just wondering about true parameter -- seems to be related to IQv2. Do we need to do something about this (I believe yes from a quick look into the code) -- can we write a proper test, that IQv2 is disabled, and does not crash anything if one tries to use it, and provides a good error message?

...ava/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeaders.java

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

mjsax · 2026-02-19T23:23:47Z

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

+    /**
+     * Generic iterator adapter for range/all operations
+     */
+    private static class TimestampedToHeadersIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {


Why are we nesting this as private classes? For the ts/kv-adapter we have it as top level classes, and re-use them for other store types, too. Can't we re-use them for other store types for the header case?

For the ts/kv-adapter we have it as top level classes

I see private static class KeyValueToTimestampedKeyValueAdapterIterator inside KeyValueToTimestampedKeyValueByteStoreAdapter

There is two classes with very similar names... KeyValueToTimestampedKeyValueIteratorAdapter (is a standalone class), and there is also KeyValueToTimestampedKeyValueAdapterIterator which is the private one... (we would rename the private one as a pointed out on some other comment to get better names -- very easy to mix up)

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

mjsax · 2026-02-19T23:30:26Z

...src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java

+
+        private final RocksDbIterator rocksDbIterator;
+
+        public TimestampedToHeadersRocksIteratorAdapter(final RocksDbIterator rocksDbIterator) {


This naming is much better than the one from KeyValueToTimestampedKeyValueByteStoreAdapter.KeyValueToTimestampedKeyValueAdapterIterator -- can we update the exiting name over there to align naming schema (best in a follow up PR)

mjsax · 2026-02-19T23:36:18Z

...rc/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java

-        return returnTimestampedStore ?
-            new RocksDBTimestampedStore(name, metricsScope()) :
-            new RocksDBStore(name, metricsScope());
+        if (returnHeadersStore) {


Suggested change

if (returnHeadersStore) {

if (returnTimestampedStore && returnHeadersStore) {

Maybe somewhat redundant, but this is how the flags are used... seems cleaner?

Well, it's redundant because returnTimestampedStore == false and returnHeadersStore == true is not something we support. -- And returnHeadersStore == true implies that returnTimestampedStore must be true as well. -- It still feel "cleaner" to check both flags?

mjsax · 2026-02-20T19:34:07Z

Build failed

aliehsaeedii added 2 commits February 11, 2026 18:08

add builders

a770815

add utest

ae0c047

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

frankvicky added the ci-approved label Feb 12, 2026

aliehsaeedii added 2 commits February 18, 2026 17:30

Merge branch 'trunk' into builders

d5b8b34

remove extra }

adeb895

github-actions bot added the needs-attention label Feb 19, 2026

aliehsaeedii added 5 commits February 19, 2026 14:04

Merge branch 'trunk' into builders

f2d589c

Merge branch 'trunk' into builders

614a4df

fix checkstyle

816ee24

add integration test

46f944e

add builder method to Stores + fix checkstyle errors

595eeb6

frankvicky reviewed Feb 19, 2026

View reviewed changes

address reviews

0458bde

mjsax reviewed Feb 19, 2026

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/state/Stores.java Outdated Show resolved Hide resolved

mjsax reviewed Feb 19, 2026

View reviewed changes

github-actions bot removed needs-attention triage PRs from the community labels Feb 20, 2026

address reviews

8fb760f

aliehsaeedii added 2 commits February 20, 2026 21:43

address review comments3

32e6a52

Merge branch 'trunk' into builders

f56c957

aliehsaeedii mentioned this pull request Feb 20, 2026

KAFKA-20132: Implement TimestampedKeyValueStoreWithHeaders (N/N) #21529

Open

aliehsaeedii added 2 commits February 20, 2026 22:46

fix javadoc

e937c3d

fix RocksDBKeyValueBytesStoreSupplier.get if/else logic

c218ad4

		* {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders} and
		* {@link org.apache.kafka.streams.state.TimestampedKeyValueStore}.

		import static org.hamcrest.CoreMatchers.equalTo;
		import static org.hamcrest.MatcherAssert.assertThat;


		private final RocksDbIterator rocksDbIterator;

		public TimestampedToHeadersRocksIteratorAdapter(final RocksDbIterator rocksDbIterator) {

	if (returnHeadersStore) {
	if (returnTimestampedStore && returnHeadersStore) {

Comments

Conversation

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

Uh oh!

Uh oh!

github-actions bot commented Feb 19, 2026

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!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliehsaeedii Feb 20, 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!

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!

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!

mjsax Feb 20, 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!

mjsax commented Feb 20, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

aliehsaeedii commented Feb 11, 2026 •

edited by github-actions bot

Loading

aliehsaeedii Feb 20, 2026 •

edited

Loading

mjsax Feb 20, 2026 •

edited

Loading