KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) by frankvicky · Pull Request #21581 · apache/kafka

frankvicky · 2026-02-25T20:12:39Z

This PR implements the upgrade integration tests for
TimestampedWindowStateStoreWithHeaders introduced in KIP-1271.

The class should be reviewd: HeadersStoreUpgradeIntegrationTest

This should not be merged before #21497

Reviewers: Matthias J. Sax matthias@confluent.io, Alieh Saeedi
asaeedi@confluent.io

aliehsaeedii

aliehsaeedii · 2026-02-26T08:18:24Z

    public void put(final K key, final ValueTimestampHeaders<V> value, final long windowStartTimestamp) {
        Objects.requireNonNull(key, "key cannot be null");
-        final Headers headers = value.headers() == null ? new RecordHeaders() : value.headers();
+        final Headers headers = value == null || value.headers() == null ? new RecordHeaders() : value.headers();


Should we do that in other metered classes as well? Did you @frankvicky notice any issue when testing here?

I introduced this change when I was writing migration test. iterator will fail due to value is null.

aliehsaeedii · 2026-02-26T08:36:07Z

+     * Iterator adapter for WindowStoreIterator that converts timestamp-only values
+     * to timestamp-with-headers format by adding empty headers.
+     */
+    private static class TimestampedToHeadersWindowStoreIteratorAdapter implements WindowStoreIterator<byte[]> {


naming: I 'm not sure but maybe TimestampedWindowToHeadersWindowStoreIteratorAdapter?
this name is longer but more descriptive.

If you compare this iterator with WindowToTimestampedWindowIteratorAdapter, should it extend TimestampedToHeadersIteratorAdapter?

aliehsaeedii · 2026-02-26T08:50:04Z

+    @Override
+    public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final long timeFrom, final long timeTo) {
+        return new TimestampedToHeadersIteratorAdapter<>(store.backwardFetchAll(timeFrom, timeTo));
+    }


What about public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {...}?

Please verify that all relevant methods have Instant-based overloads, consistent with WindowToTimestampedWindowByteStoreAdapter.java.

aliehsaeedii · 2026-02-26T08:51:51Z

+    public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom, final long timeTo) {
+        return new TimestampedToHeadersIteratorAdapter<>(store.fetchAll(timeFrom, timeTo));
+    }
+


What about public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {...}?

aliehsaeedii · 2026-02-26T09:26:38Z

+    }
+
+    /**
+     * Tests migration from TimestampedWindowStore (v2) to TimestampedWindowStoreWithHeaders (v3).


What are v2 and v3?

aliehsaeedii · 2026-02-26T09:26:53Z

+     * 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 v2 store


aliehsaeedii · 2026-02-26T09:27:59Z

+        final StreamsBuilder newBuilder = new StreamsBuilder();
+        newBuilder.addStateStore(
+                Stores.timestampedWindowStoreWithHeadersBuilder(
+                    Stores.persistentTimestampedWindowStore(WINDOW_STORE_NAME, Duration.ofMillis(RETENTION_MS), Duration.ofMillis(WINDOW_SIZE_MS), false),  // v2 supplier!


aliehsaeedii · 2026-02-26T09:28:33Z

+        kafkaStreams.close();
+        kafkaStreams = null;
+
+        // Restart with v3 builder BUT v2 supplier (proxy/adapter mode)


I'm not sure if we ever had this terminology (v2, v3`).

aliehsaeedii · 2026-02-26T09:29:12Z

+
+    @Test
+    public void shouldMigrateTimestampedWindowStoreToTimestampedWindowStoreWithHeaders() throws Exception {
+        // Phase 1: Run with TimestampedWindowStore (v2)


Didn't we check this upgrade in HeadersUpgradeIntegrationTest?

aliehsaeedii · 2026-02-26T09:32:37Z

            store.put(record.key(), ValueTimestampHeaders.make(record.value(), record.timestamp(), record.headers()));
        }
    }
+


Should we add downgrade tests as well?

mjsax · 2026-03-03T07:44:41Z

+    }
+
+    @Test
+    public void shouldFailDowngradeFromTimestampedWindowStoreWithHeadersToTimestampedWindowStore() throws Exception {


Did you look into StoreUpgradIntegrationTest#shouldFailDowngradeFromTimestampedToRegularKeyValueStore

Might be simpler to follow this pattern?

IIUC, we could not follow the same pattern since the exception has been swallowed.

kafka/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java

Lines 139 to 141 in 7ed35ab

} catch (final Exception ex) {

// ignore

}

😲 -- But it this correct to just swallow the exception???

Did some digging and it's like this since 2016, when this code was added... (#2166 -> add Segments.java -> was later refactored and the code was move into AbstractSegments.java)

Should we raise a ticket for this?

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.

mjsax · 2026-03-04T03:18:55Z

+            }
+            // Otherwise (null or corrupted data), the downgrade failed as expected
+        } catch (final Exception e) {
+            // Exception during read is expected - indicates format mismatch


We should verify the expected error message, to avoid that we catch some other exception which is unrelated to the test.

mjsax · 2026-03-04T03:22:18Z

+            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)) {


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?

mjsax · 2026-03-05T03:06:46Z

There is failing test -- was the change to AbstractSegment not correct after all?

mjsax · 2026-03-05T03:37:09Z

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: #21636

mjsax · 2026-03-06T06:38:01Z

We can rebase this PR -- feel free to merge if CI passes after rebasing.

# Conflicts: # streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java

…e#21581) This PR implements the upgrade integration tests for `TimestampedWindowStateStoreWithHeaders` introduced in KIP-1271. The class should be reviewd: `HeadersStoreUpgradeIntegrationTest` This should not be merged before apache#21497 Reviewers: Matthias J. Sax <matthias@confluent.io>, Alieh Saeedi <asaeedi@confluent.io>

github-actions Bot added triage PRs from the community streams labels Feb 25, 2026

frankvicky force-pushed the KAFKA-20134-6 branch 2 times, most recently from e5ecbba to 62aefce Compare February 26, 2026 09:29

aliehsaeedii reviewed Feb 26, 2026

View reviewed changes

frankvicky force-pushed the KAFKA-20134-6 branch from f989100 to c6ebefa Compare February 26, 2026 17:30

aliehsaeedii approved these changes Feb 26, 2026

View reviewed changes

github-actions Bot removed the triage PRs from the community label Feb 27, 2026

mjsax reviewed Mar 3, 2026

View reviewed changes

frankvicky force-pushed the KAFKA-20134-6 branch from c6ebefa to 4b6a1a8 Compare March 3, 2026 16:07

github-actions Bot added the tests Test fixes (including flaky tests) label Mar 3, 2026

mjsax reviewed Mar 4, 2026

View reviewed changes

mjsax approved these changes Mar 5, 2026

View reviewed changes

mjsax added the kip Requires or implements a KIP label Mar 6, 2026

frankvicky added 10 commits March 6, 2026 16:50

add tests

6ef0938

replace full qualified name with import

8003120

address comments

707af2a

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N)

490895e

address comments

668494a

address comments

33a154e

address comments

a654dae

address comments

0f7e43d

address comments

7c295c5

# Conflicts: # streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractSegments.java

apply spotless

96aca35

frankvicky force-pushed the KAFKA-20134-6 branch from 283fec5 to 96aca35 Compare March 6, 2026 17:00

frankvicky merged commit 5fc5d75 into apache:trunk Mar 6, 2026
23 checks passed

Conversation

frankvicky commented Feb 25, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aliehsaeedii left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Mar 5, 2026

Uh oh!

mjsax commented Mar 5, 2026

Uh oh!

mjsax commented Mar 6, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

frankvicky commented Feb 25, 2026 •

edited by github-actions Bot

Loading