You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "frankvicky (via GitHub)" <gi...@apache.org> on 2026/02/25 20:12:42 UTC
[PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
frankvicky opened a new pull request, #21581:
URL: https://github.com/apache/kafka/pull/21581
This PR implements the upgrade integration tests for
`TimestampedWindowStateStoreWithHeaders` introduced in KIP-1271.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21581:
URL: https://github.com/apache/kafka/pull/21581#issuecomment-4001807406
There is failing test -- was the change to `AbstractSegment` not correct after all?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21581:
URL: https://github.com/apache/kafka/pull/21581#issuecomment-4001896835
I looked into this, and think the tests are actually incorrect -- but it was never detected because we swallow the exception... Did PR to change this: https://github.com/apache/kafka/pull/21636
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky merged PR #21581:
URL: https://github.com/apache/kafka/pull/21581
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2881581820
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,456 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final Properties props = props();
+ final long baseTime = setupAndPopulateWindowStoreWithHeaders(props, singletonList(KeyValue.pair("key1", 100L)));
+ kafkaStreams = null;
+
+ // Attempt to downgrade to non-headers window store
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ try {
+ kafkaStreams.start();
+
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // If we can read the data correctly, the test should fail
+ if (result != null && result.value().equals("value1") && result.timestamp() == (baseTime + 100)) {
Review Comment:
It seems we have two code path that make the test pass? We reach this point, but "Fail" to verify the data, vs we crash with an `Exception`. Wondering why we need both? Would we not expect the test to either _always_ crash with and exception (and passed), _or always_ not crash but reach this code and just fail to verify?
If not, what is the race condition which influence which code path this test takes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2881574411
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,456 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final Properties props = props();
+ final long baseTime = setupAndPopulateWindowStoreWithHeaders(props, singletonList(KeyValue.pair("key1", 100L)));
+ kafkaStreams = null;
+
+ // Attempt to downgrade to non-headers window store
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ try {
+ kafkaStreams.start();
+
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // If we can read the data correctly, the test should fail
+ if (result != null && result.value().equals("value1") && result.timestamp() == (baseTime + 100)) {
+ throw new AssertionError("Expected data corruption or read failure when downgrading without cleanup, " +
+ "but data was read correctly (value=" + result.value() + ", timestamp=" + result.timestamp() + "). " +
+ "Downgrades without cleanup should not succeed.");
+ }
+ // Otherwise (null or corrupted data), the downgrade failed as expected
+ } catch (final Exception e) {
+ // Exception during read is expected - indicates format mismatch
Review Comment:
We should verify the expected error message, to avoid that we catch some other exception which is unrelated to the test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2884705172
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
I suppose we should throw the `ProcessorStateException` here. In this way, we could align the behavior between kv store and window store and also simpify the downgrade test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2881562210
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
Did some digging and it's like this since 2016, when this code was added... (https://github.com/apache/kafka/pull/2166 -> add `Segments.java` -> was later refactored and the code was move into `AbstractSegments.java`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2881571041
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,456 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final Properties props = props();
+ final long baseTime = setupAndPopulateWindowStoreWithHeaders(props, singletonList(KeyValue.pair("key1", 100L)));
+ kafkaStreams = null;
+
+ // Attempt to downgrade to non-headers window store
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ try {
+ kafkaStreams.start();
+
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // If we can read the data correctly, the test should fail
+ if (result != null && result.value().equals("value1") && result.timestamp() == (baseTime + 100)) {
+ throw new AssertionError("Expected data corruption or read failure when downgrading without cleanup, " +
+ "but data was read correctly (value=" + result.value() + ", timestamp=" + result.timestamp() + "). " +
+ "Downgrades without cleanup should not succeed.");
+ }
+ // Otherwise (null or corrupted data), the downgrade failed as expected
Review Comment:
We should never reach this part right? We either crash with an exception, or we throw the assertion-error?
So we should also fail the test here explicitly, to avoid what we pass incorrectly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2881571041
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,456 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final Properties props = props();
+ final long baseTime = setupAndPopulateWindowStoreWithHeaders(props, singletonList(KeyValue.pair("key1", 100L)));
+ kafkaStreams = null;
+
+ // Attempt to downgrade to non-headers window store
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ try {
+ kafkaStreams.start();
+
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // If we can read the data correctly, the test should fail
+ if (result != null && result.value().equals("value1") && result.timestamp() == (baseTime + 100)) {
+ throw new AssertionError("Expected data corruption or read failure when downgrading without cleanup, " +
+ "but data was read correctly (value=" + result.value() + ", timestamp=" + result.timestamp() + "). " +
+ "Downgrades without cleanup should not succeed.");
+ }
+ // Otherwise (null or corrupted data), the downgrade failed as expected
Review Comment:
We should never reach this part right? We either crash with an exception, or we throw the assertion-error?
So we should also fail the test here explicitly, to avoid what we pass incorrectly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2880558970
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
😲 -- But it this correct to just swallow the exception???
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21581:
URL: https://github.com/apache/kafka/pull/21581#issuecomment-4009902527
We can rebase this PR -- feel free to merge if CI passes after rebasing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2883168978
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
Should we raise a ticket for this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2876673340
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final StreamsBuilder headersBuilder = new StreamsBuilder();
+ headersBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(headersBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "test".getBytes());
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair("key1", "value1")),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
+ headers,
+ baseTime + 100,
+ false);
+
+ // Wait for processing
+ Thread.sleep(3000);
Review Comment:
This scrams "I am flaky" -- can we do this differently?
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
Did you look into `StoreUpgradIntegrationTest#shouldFailDowngradeFromTimestampedToRegularKeyValueStore`
Might be simpler to follow this pattern?
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final StreamsBuilder headersBuilder = new StreamsBuilder();
+ headersBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(headersBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "test".getBytes());
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair("key1", "value1")),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
+ headers,
+ baseTime + 100,
+ false);
+
+ // Wait for processing
+ Thread.sleep(3000);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ // Note: Window stores use segment-level versioning, not column family detection like key-value stores.
+ // Therefore, they do not throw an explicit "downgrade not supported" exception during initialization.
+ // Instead, attempting to read headers-aware data with a non-headers-aware store will result in
+ // data corruption (misinterpreting [headers][timestamp][value] as [timestamp][value]).
+ //
+ // This test verifies that downgrade without cleanup leads to data corruption or read failures.
+ boolean dataIsCorruptedOrUnreadable = false;
+
+ try {
+ kafkaStreams.start();
+ Thread.sleep(5000);
+
+ // Try to read the data that was written with headers
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ fail("Store should be available after startup");
+ }
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // The data should either:
+ // 1. Be null (unable to read due to format mismatch), or
+ // 2. Be corrupted (wrong value/timestamp due to misinterpreting the format)
+ if (result == null) {
+ // Data is null - indicates read failure due to format mismatch (expected)
+ dataIsCorruptedOrUnreadable = true;
+ } else {
+ // If we can read something, verify it's corrupted
+ final boolean isCorrupted = !result.value().equals("value1") || result.timestamp() != (baseTime + 100);
+ if (isCorrupted) {
+ // Data is corrupted as expected
+ dataIsCorruptedOrUnreadable = true;
+ } else {
+ // Data was read correctly - this should NOT happen
+ fail("Expected data corruption or read failure when downgrading without cleanup, " +
+ "but data was read correctly (value=" + result.value() + ", timestamp=" + result.timestamp() + "). " +
+ "downgrades without cleanup should not succeed.");
+ }
+ }
+ } catch (final Exception e) {
+ // An exception during read is also acceptable - indicates format mismatch caused a failure
+ dataIsCorruptedOrUnreadable = true;
+ } finally {
+ kafkaStreams.close(Duration.ofSeconds(30L));
+ }
+
+ // Verify that the downgrade resulted in data corruption or read failure
+ if (!dataIsCorruptedOrUnreadable) {
+ fail("Downgrade without cleanup should result in data corruption or read failure");
+ }
+ }
+
+ @Test
+ public void shouldSuccessfullyDowngradeFromTimestampedWindowStoreWithHeadersAfterCleanup() throws Exception {
+ final StreamsBuilder headersBuilder = new StreamsBuilder();
+ headersBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(headersBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "test".getBytes());
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair("key1", "value1")),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
+ headers,
+ baseTime + 100,
+ false);
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair("key2", "value2")),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
+ headers,
+ baseTime + 200,
+ false);
+
+ // Wait for processing
+ Thread.sleep(3000);
Review Comment:
We should **_never_** put `sleep()` into a test.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
+ final StreamsBuilder headersBuilder = new StreamsBuilder();
+ headersBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(headersBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "test".getBytes());
+
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair("key1", "value1")),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(), StringSerializer.class, StringSerializer.class),
+ headers,
+ baseTime + 100,
+ false);
+
+ // Wait for processing
+ Thread.sleep(3000);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder downgradedBuilder = new StreamsBuilder();
+ downgradedBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME,
+ Duration.ofMillis(RETENTION_MS),
+ Duration.ofMillis(WINDOW_SIZE_MS),
+ false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(downgradedBuilder.build(), props);
+
+ // Note: Window stores use segment-level versioning, not column family detection like key-value stores.
+ // Therefore, they do not throw an explicit "downgrade not supported" exception during initialization.
+ // Instead, attempting to read headers-aware data with a non-headers-aware store will result in
+ // data corruption (misinterpreting [headers][timestamp][value] as [timestamp][value]).
+ //
+ // This test verifies that downgrade without cleanup leads to data corruption or read failures.
+ boolean dataIsCorruptedOrUnreadable = false;
+
+ try {
+ kafkaStreams.start();
+ Thread.sleep(5000);
+
+ // Try to read the data that was written with headers
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ fail("Store should be available after startup");
+ }
+
+ final long windowStart = baseTime + 100 - ((baseTime + 100) % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch("key1", windowStart);
+
+ // The data should either:
+ // 1. Be null (unable to read due to format mismatch), or
+ // 2. Be corrupted (wrong value/timestamp due to misinterpreting the format)
+ if (result == null) {
+ // Data is null - indicates read failure due to format mismatch (expected)
+ dataIsCorruptedOrUnreadable = true;
+ } else {
+ // If we can read something, verify it's corrupted
+ final boolean isCorrupted = !result.value().equals("value1") || result.timestamp() != (baseTime + 100);
+ if (isCorrupted) {
+ // Data is corrupted as expected
+ dataIsCorruptedOrUnreadable = true;
+ } else {
+ // Data was read correctly - this should NOT happen
+ fail("Expected data corruption or read failure when downgrading without cleanup, " +
+ "but data was read correctly (value=" + result.value() + ", timestamp=" + result.timestamp() + "). " +
+ "downgrades without cleanup should not succeed.");
+ }
+ }
+ } catch (final Exception e) {
+ // An exception during read is also acceptable - indicates format mismatch caused a failure
+ dataIsCorruptedOrUnreadable = true;
+ } finally {
+ kafkaStreams.close(Duration.ofSeconds(30L));
+ }
+
+ // Verify that the downgrade resulted in data corruption or read failure
+ if (!dataIsCorruptedOrUnreadable) {
+ fail("Downgrade without cleanup should result in data corruption or read failure");
+ }
+ }
+
+ @Test
+ public void shouldSuccessfullyDowngradeFromTimestampedWindowStoreWithHeadersAfterCleanup() throws Exception {
Review Comment:
Same. Maybe compare `StoreUpgradeIntegrationTest` to simplify this code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #21581:
URL: https://github.com/apache/kafka/pull/21581#discussion_r2879125345
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/HeadersStoreUpgradeIntegrationTest.java:
##########
@@ -349,4 +363,464 @@ public void process(final Record<String, String> record) {
store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
}
}
+
+ @Test
+ public void shouldMigrateInMemoryTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(false);
+ }
+
+ @Test
+ public void shouldMigratePersistentTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(true);
+ }
+
+ /**
+ * Tests migration from TimestampedWindowStore to TimestampedWindowStoreWithHeaders.
+ * This is a true migration where both supplier and builder are upgraded.
+ */
+ private void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders(final boolean persistentStore) throws Exception {
+ // Phase 1: Run with old TimestampedWindowStore
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ persistentStore
+ ? Stores.persistentTimestampedWindowStoreWithHeaders(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false)
+ : Stores.inMemoryWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final Headers headers = new RecordHeaders();
+ headers.add("source", "migration-test".getBytes());
+ headers.add("version", "1.0".getBytes());
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, headers);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, headers);
+
+ kafkaStreams.close();
+ }
+
+ @Test
+ public void shouldProxyTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+ final StreamsBuilder oldBuilder = new StreamsBuilder();
+ oldBuilder.addStateStore(
+ Stores.timestampedWindowStoreBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedProcessor::new, WINDOW_STORE_NAME);
+
+ final Properties props = props();
+ kafkaStreams = new KafkaStreams(oldBuilder.build(), props);
+ kafkaStreams.start();
+
+ final long baseTime = CLUSTER.time.milliseconds();
+ processWindowedKeyValueAndVerifyTimestamped("key1", "value1", baseTime + 100);
+ processWindowedKeyValueAndVerifyTimestamped("key2", "value2", baseTime + 200);
+ processWindowedKeyValueAndVerifyTimestamped("key3", "value3", baseTime + 300);
+
+ kafkaStreams.close();
+ kafkaStreams = null;
+
+ // Restart with headers-aware builder but non-headers supplier (proxy/adapter mode)
+ final StreamsBuilder newBuilder = new StreamsBuilder();
+ newBuilder.addStateStore(
+ Stores.timestampedWindowStoreWithHeadersBuilder(
+ Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false), // non-headers supplier!
+ Serdes.String(),
+ Serdes.String()))
+ .stream(inputStream, Consumed.with(Serdes.String(), Serdes.String()))
+ .process(TimestampedWindowedWithHeadersProcessor::new, WINDOW_STORE_NAME);
+
+ kafkaStreams = new KafkaStreams(newBuilder.build(), props);
+ kafkaStreams.start();
+
+ verifyWindowValueWithEmptyHeaders("key1", "value1", baseTime + 100);
+ verifyWindowValueWithEmptyHeaders("key2", "value2", baseTime + 200);
+ verifyWindowValueWithEmptyHeaders("key3", "value3", baseTime + 300);
+
+ final RecordHeaders headers = new RecordHeaders();
+ headers.add("source", "proxy-test".getBytes());
+
+ // In proxy mode, headers are stripped when writing to non-headers store
+ // So we expect empty headers when reading back
+ final Headers expectedHeaders = new RecordHeaders();
+
+ processWindowedKeyValueWithHeadersAndVerify("key3", "value3-updated", baseTime + 350, headers, expectedHeaders);
+ processWindowedKeyValueWithHeadersAndVerify("key4", "value4", baseTime + 400, headers, expectedHeaders);
+
+ kafkaStreams.close();
+ }
+
+ private void processWindowedKeyValueAndVerifyTimestamped(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueAndTimestamp<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.timestampedWindowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+ final ValueAndTimestamp<String> result = store.fetch(key, windowStart);
+
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp;
+ } catch (final Exception e) {
+ return false;
+ }
+ }, 60_000L, "Could not verify timestamped value in time.");
+ }
+
+ private void processWindowedKeyValueWithHeadersAndVerify(final String key,
+ final String value,
+ final long timestamp,
+ final Headers headers,
+ final Headers expectedHeaders) throws Exception {
+ IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
+ inputStream,
+ singletonList(KeyValue.pair(key, value)),
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
+ StringSerializer.class,
+ StringSerializer.class),
+ headers,
+ timestamp,
+ false);
+
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ return result != null
+ && result.value().equals(value)
+ && result.timestamp() == timestamp
+ && result.headers().equals(expectedHeaders);
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify windowed value with headers in time.");
+ }
+
+ private void verifyWindowValueWithEmptyHeaders(final String key,
+ final String value,
+ final long timestamp) throws Exception {
+ TestUtils.waitForCondition(() -> {
+ try {
+ final ReadOnlyWindowStore<String, ValueTimestampHeaders<String>> store =
+ IntegrationTestUtils.getStore(WINDOW_STORE_NAME, kafkaStreams, QueryableStoreTypes.windowStore());
+
+ if (store == null) {
+ return false;
+ }
+
+ final long windowStart = timestamp - (timestamp % WINDOW_SIZE_MS);
+
+ final List<KeyValue<Windowed<String>, ValueTimestampHeaders<String>>> results = new LinkedList<>();
+ try (final KeyValueIterator<Windowed<String>, ValueTimestampHeaders<String>> iterator = store.all()) {
+ while (iterator.hasNext()) {
+ final KeyValue<Windowed<String>, ValueTimestampHeaders<String>> kv = iterator.next();
+ if (kv.key.key().equals(key) && kv.key.window().start() == windowStart) {
+ results.add(kv);
+ }
+ }
+ }
+
+ if (results.isEmpty()) {
+ return false;
+ }
+
+ final ValueTimestampHeaders<String> result = results.get(0).value;
+ assertNotNull(result, "Result should not be null");
+ assertEquals(value, result.value(), "Value should match");
+ assertEquals(timestamp, result.timestamp(), "Timestamp should match");
+
+ // Verify headers exist but are empty (migrated from timestamped store without headers)
+ assertNotNull(result.headers(), "Headers should not be null for migrated data");
+ assertEquals(0, result.headers().toArray().length, "Headers should be empty for migrated data");
+
+ return true;
+ } catch (final Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }, 60_000L, "Could not verify legacy value with empty headers in time.");
+ }
+
+ /**
+ * Processor for TimestampedWindowStore (without headers).
+ */
+ private static class TimestampedWindowedProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStore<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(), ValueAndTimestamp.make(record.value(), record.timestamp()), windowStart);
+ }
+ }
+
+ /**
+ * Processor for TimestampedWindowStoreWithHeaders (with headers).
+ */
+ private static class TimestampedWindowedWithHeadersProcessor implements Processor<String, String, Void, Void> {
+ private TimestampedWindowStoreWithHeaders<String, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Void, Void> context) {
+ store = context.getStateStore(WINDOW_STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<String, String> record) {
+ final long windowStart = record.timestamp() - (record.timestamp() % WINDOW_SIZE_MS);
+ store.put(record.key(),
+ ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()),
+ windowStart);
+ }
+ }
+
+ @Test
+ public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {
Review Comment:
IIUC, we could not follow the same pattern since the exception has been swallowed.
https://github.com/apache/kafka/blob/7ed35ab26b773f29f4c45210bf088f6065e1bfeb/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java#L139-L141
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org