KAFKA-20194: Add sesssion-header-store support to TDD by mjsax · Pull Request #21956 · apache/kafka

mjsax · 2026-04-03T22:44:25Z

Adds missing method TopologyTestDriver.getSessionStoreWithHeader(), and
updates TopologyTestDriverTest for all newly added header store types.

Reviewers: TengYao Chi frankvicky@apache.org, Alieh Saeedi
asaeedi@confluent.io

mjsax · 2026-04-03T22:45:10Z

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

        if (store instanceof TimestampedKeyValueStoreWithHeaders) {
            if (queryableStoreType instanceof QueryableStoreTypes.KeyValueStoreType) {
-                return (List<T>) Collections.singletonList(new GenericReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStoreWithHeaders<Object, Object>) store, ValueConverters.extractValueFromHeaders()));
+                return (List<T>) Collections.singletonList(new GenericReadOnlyKeyValueStoreFacade<>((TimestampedKeyValueStoreWithHeaders<?, ?>) store, ValueConverters.extractValueFromHeaders()));


Unrelated side cleanup -- same for StreamThreadStateStoreProvider below.

mjsax · 2026-04-03T22:45:39Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

     * @see #getTimestampedKeyValueStore(String)
-     * @see #getVersionedKeyValueStore(String)
     * @see #getTimestampedKeyValueStoreWithHeaders(String)
+     * @see #getVersionedKeyValueStore(String)


Just fixing the grouping of methods... Similar below.

mjsax · 2026-04-03T22:46:34Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java


    /**
-     * Get the {@link VersionedKeyValueStore} with the given name.
+     * Get the {@link TimestampedKeyValueStoreWithHeaders} with the given name.


This diff is weird -- I moved method getTimestampedKeyValueStoreWithHeaders before getVersionedKeyValueStore to fix the grouping in the code, too.

mjsax · 2026-04-03T22:47:01Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

     * @see #getTimestampedKeyValueStoreWithHeaders(String)
+     * @see #getVersionedKeyValueStore(String)
     * @see #getWindowStore(String)
+     * @see #getTimestampedWindowStoreWithHeaders(String)


additional fix -- this line was missing

mjsax · 2026-04-03T22:47:22Z

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

-        topology.addSource(sourceName1, Serdes.Long().deserializer(), Serdes.String().deserializer(), SOURCE_TOPIC_1);
-        topology.addSource(sourceName2, Serdes.Integer().deserializer(), Serdes.Double().deserializer(), SOURCE_TOPIC_2);
+        topology.addSource(sourceName1, new LongDeserializer(), new StringDeserializer(), SOURCE_TOPIC_1);
+        topology.addSource(sourceName2, new IntegerDeserializer(), new DoubleDeserializer(), SOURCE_TOPIC_2);


Side cleanup (similar elsewhere)

mjsax · 2026-04-03T22:49:00Z

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

        }
    }

+    // CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each eimte


IntelliJ told me to refactor, and I accepted it, but it broke the test... a lambda is a singleton, which is not allowed for a Supplier. -- So Adding this comment to prevent other to make the same mistake.

Cool...
I think it might be optimized by JVM if we use lambda.(use the same instance)

mjsax · 2026-04-03T22:49:17Z

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

-                    Stores.inMemoryKeyValueStore(keyValueStoreName),
+                persistent
+                    ? Stores.persistentKeyValueStore(keyValueStoreName)
+                    : Stores.inMemoryKeyValueStore(keyValueStoreName),


Formatting improvement.

mjsax · 2026-04-03T22:50:10Z

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

            ),
            "processor");
        topology.addStateStore(
-            persistent ?


Side cleanup / simplification. We can move the persistent check one level down, and re-use the same code for the outer "builder" call. (Similar below)

mjsax · 2026-04-04T02:35:17Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

        stateDirectory.clean();
    }

-    static class MockChangelogRegister implements ChangelogRegister {


Unused. Removing.

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

aliehsaeedii · 2026-04-07T12:53:42Z

streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java

        }
    }

+    // CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each eimte


Suggested change

// CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each eimte

// CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each time

aliehsaeedii · 2026-04-07T12:54:29Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

+     * @see #getSessionStore(String)
+     */
+    @SuppressWarnings("unchecked")
+    public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeader(final String name) {


Suggested change

public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeader(final String name) {

public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeaders(final String name) {

If yes, we should change all the java docs as well

Yes, good catch. IntelliJ "rename" feature takes care of JavaDocs automatically.

aliehsaeedii

Thanks. LGTM, just some nits. Approved anyway.

aliehsaeedii · 2026-04-07T12:55:13Z

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java

+     * @see #getSessionStore(String)
+     */
+    @SuppressWarnings("unchecked")
+    public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeader(final String name) {


If yes, we should change all the java docs as well

frankvicky

LGTM after addressing nits

Adds missing method TopologyTestDriver.getSessionStoreWithHeader(), and updates TopologyTestDriverTest for all newly added header store types. Reviewers: TengYao Chi <frankvicky@apache.org>, Alieh Saeedi <asaeedi@confluent.io>

mjsax · 2026-04-08T18:53:34Z

Merged to trunk and cherry-picked to 4.3 branch.

mjsax added streams kip Requires or implements a KIP labels Apr 3, 2026

mjsax commented Apr 3, 2026

View reviewed changes

mjsax commented Apr 4, 2026

View reviewed changes

aliehsaeedii reviewed Apr 7, 2026

View reviewed changes

streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java Show resolved Hide resolved

aliehsaeedii reviewed Apr 7, 2026

View reviewed changes

aliehsaeedii approved these changes Apr 7, 2026

View reviewed changes

frankvicky approved these changes Apr 7, 2026

View reviewed changes

mjsax force-pushed the kafka-20194-ensure-backward-compatiblity-iqV1 branch from aeb3e4f to 335ecfe Compare April 8, 2026 07:38

mjsax added 6 commits April 8, 2026 09:36

KAFKA-20194: Add sesssion-header-store support to TDD

2e2e343

checkstyle

d24f891

fix javadocs

c76e39a

checkstyle

fe2c118

review comments

7b5f389

fix previous incorrect conflict resulution during rebasing

98f234f

mjsax force-pushed the kafka-20194-ensure-backward-compatiblity-iqV1 branch from 335ecfe to 98f234f Compare April 8, 2026 16:39

mjsax merged commit 94fc9ad into apache:trunk Apr 8, 2026
23 checks passed

mjsax deleted the kafka-20194-ensure-backward-compatiblity-iqV1 branch April 8, 2026 18:47

	// CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each eimte
	// CAUTION: Do not replace with Lambda; Needs to return a new Processor instance each time

	public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeader(final String name) {
	public <K, V> SessionStoreWithHeaders<K, V> getSessionStoreWithHeaders(final String name) {

Conversation

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

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax Apr 3, 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!

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!

aliehsaeedii left a comment

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!

mjsax commented Apr 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mjsax commented Apr 3, 2026 •

edited by github-actions bot

Loading

mjsax Apr 3, 2026 •

edited

Loading