KAFKA-20329: Test headers dsl.store.format further (4/N) by aliehsaeedii · Pull Request #21827 · apache/kafka

aliehsaeedii · 2026-03-19T22:05:02Z

Parametrized 28 DSL and processor test classes to validate functionality with both headers and default store configurations. This ensures backward compatibility and correctness as Kafka Streams transitions to supporting record headers in state stores.

Modified Tests:

Materializer Tests
Windowed KStream Tests
Aggregation Tests
Cogroup Tests
Join Tests
KTable Tests

github-actions · 2026-03-27T04:01:24Z

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.

mjsax · 2026-04-06T18:29:11Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+        setUp(withHeaders);
        final StreamsBuilder builder = new StreamsBuilder();
-        final Properties props = new Properties();
-        props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);


Not sure why we did disable this explicitly as it's disabled by default -- seems we have similar stuff in other tests on this class. Should we also remove it? -- Any reason why you remove it here, but not elsewhere?

Still wondering, why we have this explicit disabling 🤔

mjsax · 2026-04-06T18:32:36Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+        setUp(withHeaders);
        final StreamsBuilder builder = new StreamsBuilder();
-        final Properties props = new Properties();
-        props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.NO_OPTIMIZATION);


Why remove this?

mjsax · 2026-04-06T18:34:51Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java

+    public void shouldNotFailIfTableIsVersionedButMaterializationIsInherited(final boolean withHeaders) {
+        setUp(withHeaders);
        final StreamsBuilder builder = new StreamsBuilder();
        final Properties props = new Properties();


setUp create it's own local Properties which is passed into TTD. So if we create new one here, that we need to pass into builder.build() below, why do we need to call setUp() at all?

This pattern of calling setUp and using local Properties is on other methods, too.

setUp must be called on every test as it used to be a @beforeEach method

mjsax · 2026-04-06T18:37:47Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java


        //should not throw an error
-        builder.build();
+        builder.build(props);


Why do we need to pass in props now, but not before? I understand that the "headers" flag is a DSL config, so if the build(...) method needs that we pass props to pick it up, it seems we need to make this change everywhere to actually enable header store? This would also apply to your other PRs (maybe even the already merged N/2? -- If we need to pass in here, all your other changes seems to actually not enable header stores?

mjsax · 2026-04-06T18:51:05Z

...a/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

+    public void shouldHaveMultipleSessionsForSameIdWhenTimestampApartBySessionGap(final EmitStrategy.StrategyType inputType, final boolean enableCaching, final boolean withHeaders) {
+        setup(inputType, enableCaching, withHeaders);
+        // This test expects caching behavior for accurate result counts
+        if (!enableCaching && !emitFinal) {


Why do we run this test only with caching disabled, and emit-on-change, but not with emit-on-final? Seems the original test setup did includ emit-on-final?

The original test considers caching as enabled:

kafka/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

Line 238 in 794d2b6

setup(inputType, true);

mjsax · 2026-04-06T18:51:21Z

...a/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

+    public void shouldHandleMultipleSessionsAndMerging(final EmitStrategy.StrategyType inputType, final boolean enableCaching, final boolean withHeaders) {
+        setup(inputType, enableCaching, withHeaders);
+        // This test expects caching behavior for accurate result counts
+        if (!enableCaching && !emitFinal) {


same question

the same as above:

kafka/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

Line 317 in 794d2b6

setup(inputType, true);

mjsax · 2026-04-06T18:57:42Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java

+            props.putAll(CONFIG);
+            props.putAll(StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String()));
+            props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, StreamsConfig.DSL_STORE_FORMAT_HEADERS);
+            builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));


Do we actually need this change at all, given that versions stores don't support headers yet?

mjsax · 2026-04-06T18:58:59Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java

+    @MethodSource("data")
+    public void testCountOfVersionedStore(final boolean withHeaders) {
+        final StreamsBuilder builder;
+        if (withHeaders) {


Same question as above

mjsax · 2026-04-06T19:07:32Z

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


-    @BeforeEach
-    public void setUp() {
+    public void setUp(final boolean withHeaders) {


Does it make sense to parameterize this test (same question for "sliding" and "windowed" case). This test already has explicit setup to test w/ and w/o header stores, for different test methods.

mjsax

Thanks for the PR. Made a pass.

frankvicky

Thanks for the PR

frankvicky · 2026-04-08T15:07:10Z

...a/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregateProcessorTest.java

    private boolean emitFinal;

-    private void setup(final EmitStrategy.StrategyType inputType, final boolean enableCaching) {
+    public static Stream<Arguments> data() {


Could we have some meaningful name?
for example emitStrategyAndCachingMatrix

frankvicky · 2026-04-08T15:07:54Z

.../test/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregateTest.java

@@ -92,14 +92,22 @@ public class KStreamSlidingWindowAggregateTest {

    public static Stream<Arguments> data() {


ditto, we should rename it.

frankvicky · 2026-04-08T15:08:23Z

...ams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java

    private boolean emitFinal;

-    public static Stream<Arguments> getEmitStrategy() {
+    public static Stream<Arguments> data() {


frankvicky · 2026-04-08T15:09:18Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableAggregateTest.java

        mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory("kafka-test").getAbsolutePath())));
-
-    private StreamsBuilder createStreamBuilderInMemory() {
+


redundant empty line

frankvicky · 2026-04-08T15:14:32Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableFilterTest.java


-    @BeforeEach
-    public void setUp() {
+    private void setupProps(final boolean withHeaders) {


nit:
I suppose we could keep the original name.

frankvicky · 2026-04-08T15:19:40Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoinTest.java

        assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
    }

+    private void setDslStoreFormat(final boolean withHeaders) {


I noticed there's Javadoc for this method in other test files. I’m fine with or without it—up to you—but the key is to stay consistent.

frankvicky · 2026-04-08T15:20:04Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableOuterJoinTest.java

        assertThat(outputTopic.readRecord(), equalTo(new TestRecord<>(expectedKey, expectedValue, null, expectedTimestamp)));
    }

+    private void setDslStoreFormat(final boolean withHeaders) {


frankvicky · 2026-04-08T15:20:10Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableKTableRightJoinTest.java

        }
    }
+
+    private void setDslStoreFormat(final boolean withHeaders) {


frankvicky · 2026-04-08T15:20:16Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapKeysTest.java

        }
    }
+
+    private void setDslStoreFormat(final boolean withHeaders) {


frankvicky · 2026-04-08T15:20:22Z

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KTableMapValuesTest.java

        }
    }
+
+    private void setDslStoreFormat(final boolean withHeaders) {


frankvicky · 2026-04-08T22:20:24Z

retrigger CI

github-actions bot added triage PRs from the community streams tests Test fixes (including flaky tests) labels Mar 19, 2026

aliehsaeedii force-pushed the dsl_headers_testing_4 branch from 6ab5990 to 9488168 Compare March 26, 2026 14:15

github-actions bot added the needs-attention label Mar 27, 2026

mjsax reviewed Apr 6, 2026

View reviewed changes

github-actions bot removed needs-attention triage PRs from the community labels Apr 7, 2026

aliehsaeedii added 2 commits April 7, 2026 23:25

utests in headers mode

eccdb52

addressed comments

162d4ad

aliehsaeedii force-pushed the dsl_headers_testing_4 branch from 9488168 to 162d4ad Compare April 7, 2026 21:29

aliehsaeedii added 2 commits April 8, 2026 00:34

address comments

0faf86e

address comments

c85ef16

frankvicky added the ci-approved label Apr 8, 2026

frankvicky reviewed Apr 8, 2026

View reviewed changes

address comments

34bbe6e

		@@ -92,14 +92,22 @@ public class KStreamSlidingWindowAggregateTest {

		public static Stream<Arguments> data() {

		mkEntry(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory("kafka-test").getAbsolutePath())));

		private StreamsBuilder createStreamBuilderInMemory() {

Conversation

aliehsaeedii commented Mar 19, 2026

Uh oh!

github-actions bot commented Mar 27, 2026

Uh oh!

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

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

aliehsaeedii Apr 7, 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!

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 left a comment

Choose a reason for hiding this comment

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!

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 commented Apr 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mjsax Apr 6, 2026 •

edited

Loading

mjsax Apr 6, 2026 •

edited

Loading

aliehsaeedii Apr 7, 2026 •

edited

Loading