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

aliehsaeedii · 2026-02-11T16:01:44Z

This PR implements the changelogging layer of the
TimestampedKeyValueStoreWithHeaders introduced in KIP-1271.

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

frankvicky

Overall LGTM

frankvicky · 2026-02-11T17:43:37Z

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

+        assertEquals(hi, collector.collected().get(0).key());
+        assertArrayEquals(there.value(), (byte[]) collector.collected().get(0).value());
+        assertEquals(97L, collector.collected().get(0).timestamp());
+        Headers headers0 = collector.collected().get(0).headers();


frankvicky · 2026-02-11T17:43:54Z

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

+        assertEquals(hi, collector.collected().get(0).key());
+        assertArrayEquals(there.value(), (byte[]) collector.collected().get(0).value());
+        assertEquals(97L, collector.collected().get(0).timestamp());
+        Headers headers0 = collector.collected().get(0).headers();


frankvicky · 2026-02-11T17:43:58Z

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

+        assertEquals(hello, collector.collected().get(1).key());
+        assertArrayEquals(world.value(), (byte[]) collector.collected().get(1).value());
+        assertEquals(98L, collector.collected().get(1).timestamp());
+        Headers headers1 = collector.collected().get(1).headers();


frankvicky · 2026-02-11T17:44:14Z

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

+        assertArrayEquals(there.value(), (byte[]) collector.collected().get(0).value());
+        assertEquals(97L, collector.collected().get(0).timestamp());
+
+        Headers headers = collector.collected().get(0).headers();


frankvicky · 2026-02-11T17:44:21Z

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

+        assertEquals(hi, collector.collected().get(0).key());
+        assertArrayEquals(there.value(), (byte[]) collector.collected().get(0).value());
+        assertEquals(97L, collector.collected().get(0).timestamp());
+        Headers headers0 = collector.collected().get(0).headers();


...ams/src/main/java/org/apache/kafka/streams/processor/internals/InternalProcessorContext.java

mjsax · 2026-02-12T00:34:01Z

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

+    /**
+     * Extract raw value from serialized ValueTimestampHeaders.
+     */
+    static byte[] rawValue(final byte[] rawValueTimestampHeaders) {


We have static <T> T value(...) above -- given that we need rawValue here, wondering if value(...) from above is actually needed or not at all?

that method returns the deserialized value while raawValue returns a byte[]

Yes, I am not questioning that we add rawValue in this PR, I am asking why did we add value(...) in a previous PR, and when would we use it?

But it's somewhat unrelated to this PR, and it's only internal code here. So we could remove value(...) if unused also at some point in the future. Was just wondering.

We can remove it at some point if not needed: https://issues.apache.org/jira/browse/KAFKA-20193

mjsax · 2026-02-12T00:37:36Z

...che/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.java

+        }
+    }
+
+    void log(final Bytes key, final byte[] value, final long timestamp, final Headers headers) {


If we just extend InternalProcessorContext.log(...) with a new Headers parameter and not add a new overload, and update all existing code accordingly, I believe we can remove the method and reuse the existing one from ChangeLoggingKeyValueBytesStore

It's a very big change. It will chnage many tests.

...kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeadersTest.java

mjsax · 2026-02-14T05:14:34Z

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

+    /**
+     * Extract raw value from serialized ValueTimestampHeaders.
+     */
+    static byte[] rawValue(final byte[] rawValueTimestampHeaders) {


Yes, I am not questioning that we add rawValue in this PR, I am asking why did we add value(...) in a previous PR, and when would we use it?

But it's somewhat unrelated to this PR, and it's only internal code here. So we could remove value(...) if unused also at some point in the future. Was just wondering.

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

mjsax · 2026-02-15T20:29:04Z

Failing unit test. Needs a fix.

mjsax · 2026-02-17T18:54:02Z

...che/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.java

+            valueTimestampHeaders == null
+                ? internalContext.recordContext().timestamp()
+                : timestamp(valueTimestampHeaders),
+            headers(valueTimestampHeaders)


Think we need to do the same as for ts, and use internalContext.recordContext().headers() if valueTimestampHeaders is null.

frankvicky

LGTM, thanks for the PR.

aliehsaeedii added 2 commits February 11, 2026 16:49

add changelogging layer

61d0419

add log method to InternalProcessorContext classes

b94eea0

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

aliehsaeedii mentioned this pull request Feb 11, 2026

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

Merged

frankvicky reviewed Feb 11, 2026

View reviewed changes

add missing final keyword

973d7c1

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

mjsax reviewed Feb 12, 2026

View reviewed changes

address reviews

4a05553

mjsax approved these changes Feb 14, 2026

View reviewed changes

mjsax reviewed Feb 14, 2026

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java Outdated Show resolved Hide resolved

frankvicky mentioned this pull request Feb 17, 2026

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (4/N) #21493

Merged

mjsax reviewed Feb 17, 2026

View reviewed changes

fix test + add null value check

d81dddb

aliehsaeedii force-pushed the changelogging branch from 550d18e to d81dddb Compare February 17, 2026 20:28

aliehsaeedii added 2 commits February 17, 2026 21:29

fix test

72497d6

Merge branch 'trunk' into changelogging

3b6ee2d

frankvicky approved these changes Feb 18, 2026

View reviewed changes

frankvicky merged commit 85c6313 into apache:trunk Feb 18, 2026
22 checks passed

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!

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!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mjsax commented Feb 15, 2026

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

aliehsaeedii commented Feb 11, 2026 •

edited by github-actions bot

Loading