KAFKA-20220: Enable TimestampedKVStoreWithHeaders in DSL (1/N) by aliehsaeedii · Pull Request #21572 · apache/kafka

aliehsaeedii · 2026-02-24T22:53:01Z

This PR introduces the DslStoreFormat enum and extends DslKeyValueParams
to enable headers-aware key-value stores in the Kafka Streams DSL,
implementing the foundational infrastructure for
KIP-1285.

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

frankvicky

Thanks for the PR

frankvicky · 2026-02-26T10:35:21Z

...rc/test/java/org/apache/kafka/streams/processor/internals/KeyValueStoreMaterializerTest.java

+    @Test
+    public void shouldNotBuildHeadersAwareStoreWithCachingEvenIfExplicitlySet() {
+        doReturn("headers")
+                .when(streamsConfig).getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
+
+        final MaterializedInternal<String, String, KeyValueStore<Bytes, byte[]>> materialized =
+            new MaterializedInternal<>(Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("store").withCachingEnabled(), nameProvider, STORE_PREFIX);
+
+        final TimestampedKeyValueStoreWithHeaders<String, String> store = getHeadersAwareStore(materialized);
+
+        final WrappedStateStore logging = (WrappedStateStore) ((WrappedStateStore) store).wrapped();
+        assertThat(store, instanceOf(MeteredTimestampedKeyValueStoreWithHeaders.class));
+        assertThat(logging, instanceOf(ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders.class));
+        // Verify that caching layer was NOT added
+        assertThat(logging, not(instanceOf(CachingKeyValueStore.class)));
+    }


It seems that the test will fail because that we only check cache condition for versioned store ?
https://github.com/aliehsaeedii/kafka/blob/560f78cd966899d21ad94501373bc421204898eb/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java#L84-L90

Thanks @frankvicky. Fixed the test. The code is correct but the test does not make sense.

mjsax

Made a pass (skipped tests for now)

mjsax · 2026-02-27T06:11:14Z

...c/main/java/org/apache/kafka/streams/kstream/internals/AbstractConfigurableStoreFactory.java

                config.originals()
            );
        }
+        if (dslStoreFormat == null) {


We did not update the constructor to pass in DslStoreFormat, so it would always be null here atm. Is this something that will change with follow up PRs? Or would we never need to pass via constructor, and this check is not necessary as we would always set it from the config?

yes we must always set it from the confg. So i remove the if clause.

mjsax · 2026-02-27T06:12:03Z

...c/main/java/org/apache/kafka/streams/kstream/internals/AbstractConfigurableStoreFactory.java

        }
+        if (dslStoreFormat == null) {
+            final String dslStoreFormatValue = config.getString(StreamsConfig.DSL_STORE_FORMAT_CONFIG);
+            if (dslStoreFormatValue.equalsIgnoreCase("HEADERS")) {


Suggested change

if (dslStoreFormatValue.equalsIgnoreCase("HEADERS")) {

if (dslStoreFormatValue.equalsIgnoreCase(StreamsConfig.DSL_STORE_FORMAT_HEADERS)) {

mjsax · 2026-02-27T06:14:48Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

+    public static final String DSL_STORE_FORMAT_DEFAULT = "DEFAULT";
+    private static final String DSL_STORE_FORMAT_DOC = "Controls the state store type used by the DSL store supplier (see " +
+        "config 'dsl.store.suppliers.class'). " +
+        "'default' uses timestamped stores. " +


This is not totally correct, right? Or do we want to just ignore this fact? Might be ok as-is; just double checking.

I changed it to

private static final String DSL_STORE_FORMAT_DOC = "Specifies the state store format for DSL operators. " + "'DEFAULT' creates either timestamped or plain state stores, depending on context. " + "'HEADERS' creates headers-aware stores that preserve record headers.";

This is more accurate.

mjsax · 2026-02-27T06:15:23Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

+    public static final String DSL_STORE_FORMAT_CONFIG = "dsl.store.format";
+    public static final String DSL_STORE_FORMAT_DEFAULT = "DEFAULT";
+    private static final String DSL_STORE_FORMAT_DOC = "Controls the state store type used by the DSL store supplier (see " +
+        "config 'dsl.store.suppliers.class'). " +


Suggested change

"config 'dsl.store.suppliers.class'). " +

"config '" + DSL_STORE_SUPPLIER_CONFIG + "'). " +

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

mjsax · 2026-02-27T06:23:15Z

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

-                    ? Stores.persistentTimestampedKeyValueStore(params.name())
-                    : Stores.persistentKeyValueStore(params.name());
+            final DslStoreFormat storeFormat = params.dslStoreFormat();
+            if (storeFormat.equals(DslStoreFormat.HEADERS)) {


Suggested change

if (storeFormat.equals(DslStoreFormat.HEADERS)) {

if (storeFormat == DslStoreFormat.HEADERS) {

Or maybe even use switch (storeFormat) ?

mjsax · 2026-02-27T06:33:26Z

...c/main/java/org/apache/kafka/streams/kstream/internals/AbstractConfigurableStoreFactory.java

+            if (dslStoreFormatValue.equalsIgnoreCase("HEADERS")) {
+                dslStoreFormat = DslStoreFormat.HEADERS;
+            } else { // DEFAULT
+                dslStoreFormat = DslStoreFormat.TIMESTAMPED;


This logic is too generic -- it works for most DSL operators but some need to translate "DEFAULT" to PLAIN.

I don't think we can make this translation here, but need to push it down a lower layer where it's know how to translate it?

mjsax · 2026-02-27T06:34:04Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java

                    materialized.keySerde(),
                    materialized.valueSerde());
-        } else {
+        } else if (storeFormat.equals(DslStoreFormat.HEADERS)) {


Suggested change

} else if (storeFormat.equals(DslStoreFormat.HEADERS)) {

} else if (storeFormat == DslStoreFormat.HEADERS) {

Similar elsewhere. Enums are not object; we can use ==

mjsax · 2026-02-27T06:35:38Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java

                    supplier,
                    materialized.keySerde(),
                    materialized.valueSerde());
+        } else if (storeFormat.equals(DslStoreFormat.PLAIN)) {


Given the current code-setup, dslStoreFormat() would never return PLAIN, right? So this case is currently not possible?

mjsax · 2026-02-27T06:36:49Z

...ms/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java

        final LeftOrRightValueSerde<V1, V2> leftOrRightValueSerde = new LeftOrRightValueSerde<>(streamJoined.valueSerde(), streamJoined.otherValueSerde());

-        final DslKeyValueParams dslKeyValueParams = new DslKeyValueParams(name, false);
+        final DslStoreFormat storeFormat = dslStoreFormat() == DslStoreFormat.HEADERS ? dslStoreFormat() : DslStoreFormat.PLAIN;


Looks not clean to me, to override TIMESTAMPED as PLAIN here?

frankvicky · 2026-02-27T10:43:00Z

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

+     * @param dslStoreFormat the format of the state store, see ({@link DslStoreFormat}
+     */
+    public DslKeyValueParams(final String name, final DslStoreFormat dslStoreFormat) {
+        Objects.requireNonNull(name);


Duplicate check

frankvicky · 2026-02-27T10:44:09Z

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

     * @param name          the name of the store (cannot be {@code null})
     * @param isTimestamped whether the returned stores should be timestamped, see ({@link TimestampedKeyValueStore}
     */
+    @Deprecated


@aliehsaeedii do you want to keep it for now?

aliehsaeedii · 2026-02-27T11:12:25Z

@frankvicky

@aliehsaeedii do you want to keep it for now?

we keep the constructor but we deprecate it as it is listed in the KIP (kip-1285).

mjsax · 2026-02-28T04:22:33Z

...c/main/java/org/apache/kafka/streams/kstream/internals/AbstractConfigurableStoreFactory.java

+        if (dslStoreFormatValue.equalsIgnoreCase(StreamsConfig.DSL_STORE_FORMAT_HEADERS)) {
+            dslStoreFormat = DslStoreFormat.HEADERS;
+        }
+        // else dslStoreFormat remains null and the lower layers decide between PLAIN and TIMESTAMPED


I actually proposed, that we would pass dslStoreFormatValue down to lower layers, and let them translated the String config from dsl.store.format to the DslStoreFormat enum...

But I guess this also work. Feel a little bit less clean to me personally, but maybe not worth to change now that this PR is already merged.

aliehsaeedii added 18 commits February 11, 2026 18:08

add builders

a770815

add utest

ae0c047

Merge branch 'trunk' into builders

d5b8b34

remove extra }

adeb895

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

address reviews

0458bde

address reviews

8fb760f

address review comments3

32e6a52

Merge branch 'trunk' into builders

f56c957

fix javadoc

e937c3d

fix RocksDBKeyValueBytesStoreSupplier.get if/else logic

c218ad4

address review comments4

ea51092

address review comments 5

91e510f

enable dsl for kv store

0aae133

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

aliehsaeedii added 6 commits February 24, 2026 23:54

un-comment deprecated things

dbb40a3

Merge branch 'trunk' into dsl_with_headers

dad2b5d

couple dslFormat and config again

3ab45bb

fix KeyValueStoreMaterializer

9bf5c85

decouple DslStoreFormat enum and the config

99e824d

should not be case-sensitive

560f78c

aliehsaeedii changed the title ~~/KAFKA-20220: Enable TimestampedKVStoreWithHeaders in DSL (1/N)~~ KAFKA-20220: Enable TimestampedKVStoreWithHeaders in DSL (1/N) Feb 25, 2026

frankvicky reviewed Feb 26, 2026

View reviewed changes

fix tests

63c70d4

frankvicky added the ci-approved label Feb 26, 2026

frankvicky approved these changes Feb 26, 2026

View reviewed changes

github-actions bot removed the triage PRs from the community label Feb 27, 2026

mjsax reviewed Feb 27, 2026

View reviewed changes

aliehsaeedii added 2 commits February 27, 2026 10:43

address review comments

9205a99

Objects.requireNonNull for name as well

1c643ed

frankvicky reviewed Feb 27, 2026

View reviewed changes

remove duplicate check

f2869b7

frankvicky approved these changes Feb 27, 2026

View reviewed changes

frankvicky merged commit 9f22628 into apache:trunk Feb 27, 2026
23 checks passed

mjsax reviewed Feb 28, 2026

View reviewed changes

mjsax added the kip Requires or implements a KIP label Mar 6, 2026

	if (dslStoreFormatValue.equalsIgnoreCase("HEADERS")) {
	if (dslStoreFormatValue.equalsIgnoreCase(StreamsConfig.DSL_STORE_FORMAT_HEADERS)) {

	"config 'dsl.store.suppliers.class'). " +
	"config '" + DSL_STORE_SUPPLIER_CONFIG + "'). " +

	if (storeFormat.equals(DslStoreFormat.HEADERS)) {
	if (storeFormat == DslStoreFormat.HEADERS) {

	} else if (storeFormat.equals(DslStoreFormat.HEADERS)) {
	} else if (storeFormat == DslStoreFormat.HEADERS) {

Conversation

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

mjsax 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!

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!

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 commented Feb 27, 2026

Uh oh!

Uh oh!

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

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

aliehsaeedii commented Feb 24, 2026 •

edited by github-actions bot

Loading

mjsax Feb 28, 2026 •

edited

Loading