KStreams PAPI test for SessionStoreWithHeaders by lucliu1108 · Pull Request #4213 · confluentinc/schema-registry

Lucy Liu (lucliu1108) · 2026-03-20T15:19:27Z

Summary

This PR adds:

SessionStoreWithHeadersIntegrationTest, which did PAPI test on sessionStoreWithHeaders, and validate store operations in the processor.

What

Checklist

Please answer the questions with Y, N or N/A if not applicable.

[ ] Contains customer facing changes? Including API/behavior changes
[ ] Is this change gated behind config(s)?
- List the config(s) needed to be set to enable this change
[ ] Did you add sufficient unit test and/or integration test coverage for this PR?
- If not, please explain why it is not required
[ ] Does this change require modifying existing system tests or adding new system tests?
- If so, include tracking information for the system test changes
[ ] Must this be released together with other change(s), either in this repo or another one?
- If so, please include the link(s) to the changes that must be released together

References

JIRA:

Test & Review

Open questions / Follow-ups

confluent-cla-assistant · 2026-03-20T15:19:41Z

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

Copilot

Pull request overview

Adds a Kafka Streams integration test suite to exercise SessionStoreWithHeaders behavior when using header-based Schema ID transport, including processor-driven store operations, IQv1 querying, and changelog tombstone header validation.

Changes:

Introduces SessionStoreWithHeadersIntegrationTest covering many session-store operations and verifying schema-id headers on produced results.
Adds an IQv1 test path using a custom QueryableStoreType for SessionStoreWithHeaders.
Updates streams-integration-tests Maven dependencies (Kafka clients/streams and test-utils).

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 9 comments.

File	Description
streams-integration-tests/src/test/java/io/confluent/kafka/streams/integration/SessionStoreWithHeadersIntegrationTest.java	New integration tests validating `SessionStoreWithHeaders` operations and header-based schema-id behavior (including IQv1 + changelog tombstones).
streams-integration-tests/pom.xml	Adjusts Kafka-related test dependencies/versions for the integration-tests module.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-03-20T15:27:31Z

+            // Consume outputs: 5 PUTs + 1 REMOVE = 6 records
+            consumeRecords(iqv1OutputTopic, "iqv1-session-consumer", 6);
+
+            // Query store via IQv1
+            ReadOnlySessionStore<GenericRecord, AggregationWithHeaders<GenericRecord>> store =
+                streams.store(
+                    StoreQueryParameters.fromNameAndType(iqv1StoreName, new SessionStoreWithHeadersType<>()));
+            assertNotNull(store, "Store should be accessible via IQv1");


In shouldVerifyIQv1Operations, the call to consumeRecords(..., 6) ignores the returned list and does not assert that the expected 6 output records were actually produced/consumed before querying the store. If the poll loop times out early, the subsequent IQv1 queries can run before all inputs are processed, making the test flaky. Consider asserting the consumed record count (or otherwise waiting for processing) before performing the store queries.

Copilot · 2026-03-20T15:27:32Z

+            Windowed<GenericRecord> sessionKey = new Windowed<>(
+                record.key(),
+                new org.apache.kafka.streams.kstream.internals.SessionWindow(record.timestamp(), record.timestamp())
+            );


Another construction of a session window via the ...internals.SessionWindow class. Consider reusing the same helper for building Windowed<GenericRecord> keys so the internal dependency is localized and easier to update if Kafka changes the internal type.

Copilot · 2026-03-20T15:27:32Z

+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams-test-utils</artifactId>
+            <version>4.3.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
        <dependency>


kafka-streams-test-utils is added as a direct dependency, but it doesn't appear to be used anywhere under streams-integration-tests/src/test/java (no imports/usages found). If it’s not required, consider removing it to keep the test classpath minimal; if it is required transitively, adding a comment explaining why would help future maintenance.

Suggested change

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-streams-test-utils</artifactId>

<version>4.3.0-SNAPSHOT</version>

<scope>test</scope>

</dependency>

<dependency>

Copilot · 2026-03-20T15:27:33Z

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
+            <version>4.3.0-SNAPSHOT</version>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.confluent.networking</groupId>
+                    <artifactId>cc-custom-dns-resolver-java18</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>4.3.0-SNAPSHOT</version>
            <classifier>test</classifier>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>io.confluent.networking</groupId>
                    <artifactId>cc-custom-dns-resolver-java18</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
+            <version>4.3.0-SNAPSHOT</version>
            <scope>test</scope>
            <exclusions>
                <exclusion>
                    <groupId>io.confluent.networking</groupId>
                    <artifactId>cc-custom-dns-resolver-java18</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams-test-utils</artifactId>
+            <version>4.3.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>


The module now pins kafka-clients, kafka-streams, and kafka-streams-test-utils to 4.3.0-SNAPSHOT, while other Kafka artifacts in the same POM (eg kafka-server-common and kafka_${kafka.scala.version}) still come from the parent-managed ${kafka.version}. Mixing Kafka versions in a single test classpath is very likely to cause dependency convergence / linkage issues at runtime. Consider using the same ${kafka.version} for all Kafka artifacts (ideally via dependencyManagement) and avoid hardcoding a SNAPSHOT here unless the build is explicitly set up to resolve it and all Kafka deps are aligned.

TengYao Chi (frankvicky) · 2026-03-31T22:17:45Z

+        }
+    }
+
+    private List<ConsumerRecord<GenericRecord, GenericRecord>> consumeRecords(


Should we add a check when received count > expected count?

Edit: added the check after poll loop

…h-header-test

Lucy Liu (lucliu1108) · 2026-06-01T22:08:48Z

Hi Alieh Saeedi (@aliehsaeedii)
I applied the fixes suggested in #4204

Must-Fix 1 (session ordering bug) — assertRangeResults no longer uses a Map; now compares index-by-index (expectedPairs[i] vs results.get(startIdx + i)), so backward-fetch tests can't pass with forward-ordered results
Must-Fix 3 (consumeRecords silent timeout) — Unified into single generic <V> consumeRecords(topic, groupId, expectedCount, Class<?> valueDeserializerClass) with strict assertEquals(expectedCount, results.size(), ...). Now covers regular and changelog calls
Should-Fix 4 (duplicate lines) — Removed createConsumerProps, createChangelogConsumerProps, consumeChangelogRecords
Per-test 11 (shouldVerifyIQv1Operations no count check) — Captured iqv1Results and added assertEquals(6, iqv1Results.size(), ...)
Should-Fix 7 (trailing newline) — Added at EOF

Lucy Liu (lucliu1108) requested a review from a team as a code owner March 20, 2026 15:19

Copilot AI review requested due to automatic review settings March 20, 2026 15:19

Copilot started reviewing on behalf of Lucy Liu (lucliu1108) March 20, 2026 15:20 View session

Copilot AI reviewed Mar 20, 2026

View reviewed changes

TengYao Chi (frankvicky) reviewed Mar 31, 2026

View reviewed changes

Lucy Liu (lucliu1108) added 3 commits April 14, 2026 15:27

add test

5f203fa

address comments

408410f

add received vs expected record check when consuming records

79a5c47

Lucy Liu (lucliu1108) force-pushed the session-store-with-header-test branch from f2e92c9 to 79a5c47 Compare April 14, 2026 20:36

Lucy Liu (lucliu1108) requested review from Alieh Saeedi (aliehsaeedii), Bill Bejeck (bbejeck), TengYao Chi (frankvicky) and Matthias J. Sax (mjsax) April 16, 2026 17:30

Merge remote-tracking branch 'upstream/master' into session-store-wit…

d9ce334

…h-header-test

fix test

5880a1a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KStreams PAPI test for SessionStoreWithHeaders#4213

KStreams PAPI test for SessionStoreWithHeaders#4213
Lucy Liu (lucliu1108) wants to merge 5 commits into
confluentinc:masterfrom
lucliu1108:session-store-with-header-test

Lucy Liu (lucliu1108) commented Mar 20, 2026

Uh oh!

confluent-cla-assistant Bot commented Mar 20, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Mar 20, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 20, 2026

Uh oh!

Copilot AI Mar 20, 2026

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 20, 2026

Uh oh!

TengYao Chi (frankvicky) Mar 31, 2026

Uh oh!

Lucy Liu (lucliu1108) Apr 14, 2026

Uh oh!

Lucy Liu (lucliu1108) commented Jun 1, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

Lucy Liu (lucliu1108) commented Mar 20, 2026

Summary

What

Checklist

References

Test & Review

Open questions / Follow-ups

Uh oh!

confluent-cla-assistant Bot commented Mar 20, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Mar 20, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 20, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Mar 20, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI Mar 20, 2026

Choose a reason for hiding this comment

Uh oh!

TengYao Chi (frankvicky) Mar 31, 2026

Choose a reason for hiding this comment

Uh oh!

Lucy Liu (lucliu1108) Apr 14, 2026

Choose a reason for hiding this comment

Uh oh!

Lucy Liu (lucliu1108) commented Jun 1, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants