KStreams PAPI test for TimestampedWindowStoreWithHeaders by lucliu1108 · Pull Request #4212 · confluentinc/schema-registry

Lucy Liu (lucliu1108) · 2026-03-20T15:08:00Z

Summary

This PR adds:

TimestampedWindowStoreWithHeadersIntegrationTest, which did PAPI test on timestampedWindowStoreWithHeaders, 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:08:13Z

🎉 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 new Streams integration test class to validate TimestampedWindowStoreWithHeaders behavior (including header-based schema ID transport) across a variety of store operations and query paths.

Changes:

Introduces TimestampedWindowStoreWithHeadersIntegrationTest covering PAPI-style store ops, IQv1 querying, delete/tombstone behavior, and iterator method behavior.
Adds helper utilities for topic creation, Streams lifecycle, producing/consuming records, and schema-id header assertions.
Fixes trailing indentation/formatting at the end of streams-integration-tests/pom.xml.

Reviewed changes

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

File	Description
`streams-integration-tests/src/test/java/io/confluent/kafka/streams/integration/TimestampedWindowStoreWithHeadersIntegrationTest.java`	New, comprehensive integration test suite for window-store-with-headers operations and IQ querying.
`streams-integration-tests/pom.xml`	Minor formatting change (project closing tag alignment).

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

Copilot · 2026-03-20T15:13:53Z

+                        "Tombstone record should have key schema ID header");
+                }
+            }
+            System.out.println("Tombstone count: " + tombstoneCount);


Avoid printing to stdout from tests. This adds noise to CI logs and can hide relevant output; if you need diagnostics, use assertions/messages or a test logger that can be enabled when debugging.

Suggested change

System.out.println("Tombstone count: " + tombstoneCount);

Copilot · 2026-03-20T15:13:54Z

+        assertTrue(startedLatch.await(30, TimeUnit.SECONDS), "KafkaStreams should reach RUNNING state");
+        return streams;


If the RUNNING-state await times out, startStreamsAndAwaitRunning throws an AssertionError after starting KafkaStreams, but the instance is never closed (call site won't assign it due to the exception). Consider closing streams before failing (e.g., in a try/finally around the await) to avoid thread/resource leaks and noisy CI failures when the startup is flaky.

Suggested change

assertTrue(startedLatch.await(30, TimeUnit.SECONDS), "KafkaStreams should reach RUNNING state");

return streams;

boolean running = false;

try {

running = startedLatch.await(30, TimeUnit.SECONDS);

assertTrue(running, "KafkaStreams should reach RUNNING state");

return streams;

} finally {

if (!running) {

closeStreams(streams);

}

}

TengYao Chi (frankvicky)

Thanks for the PR

TengYao Chi (frankvicky) · 2026-03-31T22:04:25Z

+        }
+
+        /**
+         * Fetch range for event-2 from 6-11min (240000ms to 660000ms).


seems the comment not accurate?

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

+        @Override
+        public void process(Record<GenericRecord, GenericRecord> record) {
+            String operation = record.value().get("operation").toString();
+            long windowStart = (record.timestamp() / WINDOW_SIZE.toMillis()) * WINDOW_SIZE.toMillis();


Could we have a helper method for window start/end calculation?

Edit: added helper method calculateWindowStartTime to calculate window start time

TengYao Chi (frankvicky) · 2026-03-31T22:09:52Z

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


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

Edit: added a number check after the poll loop completed

address comments address comments

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

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

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

Copilot AI reviewed Mar 20, 2026

View reviewed changes

Lucy Liu (lucliu1108) force-pushed the timestamped-windowed-store-header-test branch from 86f9cf8 to fb6a067 Compare March 20, 2026 15:42

TengYao Chi (frankvicky) reviewed Mar 31, 2026

View reviewed changes

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

add test

6a1e414

address comments

0eaa725

address comments address comments

address comments

d3c244e

Lucy Liu (lucliu1108) force-pushed the timestamped-windowed-store-header-test branch from fb6a067 to d3c244e Compare April 14, 2026 20:11

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

		assertTrue(startedLatch.await(30, TimeUnit.SECONDS), "KafkaStreams should reach RUNNING state");
		return streams;

-        assertTrue(startedLatch.await(30, TimeUnit.SECONDS), "KafkaStreams should reach RUNNING state");
-        return streams;
+        boolean running = false;
+        try {
+            running = startedLatch.await(30, TimeUnit.SECONDS);
+            assertTrue(running, "KafkaStreams should reach RUNNING state");
+            return streams;
+        } finally {
+            if (!running) {
+                closeStreams(streams);
+            }
+        }

Conversation

Lucy Liu (lucliu1108) commented Mar 20, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

TengYao Chi (frankvicky) left a comment

Choose a reason for hiding this comment

Uh oh!

TengYao Chi (frankvicky) Mar 31, 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!

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Lucy Liu (lucliu1108) commented Mar 20, 2026 •

edited

Loading