SchemaId header store DSL test for TimestampedWindowStoreWithHeader

[Lucy Liu (lucliu1108)]

Summary

Add DSL-level integration tests for TimestampedWindowStoreWithHeaders to verify that schema ID headers are correctly propagated through windowed Kafka Streams operations backed by header-aware state stores.

Each test contains

• parameterization across cache/uncached, with/without grace period senarios
• IQv1 state store queries
• Changelog header vadliation

Operations

• Count (tumbling windows) — shouldCountWithTumblingWindows, with null handling
• Aggregate (tumbling windows) — shouldAggregateWithTumblingWindows, with null handling
• Reduce (tumbling windows) — shouldReduceWithTumblingWindows
• Count (hopping windows) — shouldCountWithHoppingWindows
• Stream-Stream Join — shouldStreamStreamsJoinWithHeaders
• Suppress — shouldSuppressWithHeaders, with null handling

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]

🎉 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 DSL-level integration coverage to validate SchemaId header propagation through windowed/stateful Kafka Streams operations backed by header-aware timestamped window stores.

Changes:

• Introduces a new parameterized DSL integration test suite for TimestampedWindowStoreWithHeaders, covering count/aggregate/reduce/hopping windows, stream-stream join, and suppression.
• Removes the prior stateless header-schema-id integration test.
• Updates the streams-integration-tests Maven module dependencies (Kafka artifacts + JUnit params).

Reviewed changes

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

File	Description
streams-integration-tests/src/test/java/io/confluent/kafka/streams/integration/dsl/TimestampedWindowStoreWithHeadersDslIntegrationTest.java	New DSL integration tests validating schema-id headers via IQv1 queries and changelog consumption across cache/grace combinations.
streams-integration-tests/src/test/java/io/confluent/kafka/streams/integration/KafkaStreamsHeaderSchemaIdIntegrationTest.java	Removes the previous stateless filter-based header propagation integration test.
streams-integration-tests/pom.xml	Adds JUnit params and modifies Kafka dependency versions / adds streams test utils.

---

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

[Copilot]

This fixed sleep is used to wait for the join results, but fixed delays are a common source of flaky integration tests (too short under load, too long when fast). Prefer waiting on a condition (e.g., poll loop until 2 output records are observed or until a timeout) instead of Thread.sleep.

[Copilot]

The join output consumption does a single poll(Duration.ofSeconds(10)) and immediately asserts on the number of records. This is prone to flakes (records can arrive after the first poll). Prefer polling in a loop until you collect the expected number of records (or a deadline) similar to other tests in the repo that consume N messages.

Suggested change

	List<ConsumerRecord<GenericRecord, GenericRecord>> outputRecords = new ArrayList<>();
	try (KafkaConsumer<GenericRecord, GenericRecord> consumer = new KafkaConsumer<>(consumerProps, keySerde.deserializer(), aggSerde.deserializer())) {
	consumer.subscribe(Collections.singletonList(outputTopic));
	ConsumerRecords<GenericRecord, GenericRecord> records = consumer.poll(java.time.Duration.ofSeconds(10));
	records.forEach(outputRecords::add);
	}
	assertEquals(2, outputRecords.size(), "Should have 2 join results (kafka and streams)");
	final int expectedOutputRecordCount = 2;
	final Instant pollDeadline = Instant.now().plus(Duration.ofSeconds(10));
	List<ConsumerRecord<GenericRecord, GenericRecord>> outputRecords = new ArrayList<>();
	try (KafkaConsumer<GenericRecord, GenericRecord> consumer = new KafkaConsumer<>(consumerProps, keySerde.deserializer(), aggSerde.deserializer())) {
	consumer.subscribe(Collections.singletonList(outputTopic));
	while (outputRecords.size() < expectedOutputRecordCount && Instant.now().isBefore(pollDeadline)) {
	ConsumerRecords<GenericRecord, GenericRecord> records = consumer.poll(Duration.ofMillis(500));
	records.forEach(outputRecords::add);
	}
	}
	assertEquals(expectedOutputRecordCount, outputRecords.size(), "Should have 2 join results (kafka and streams)");

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 5 comments.

---

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