Skip to content

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (5/N)#21497

Merged
mjsax merged 12 commits intoapache:trunkfrom
frankvicky:KAFKA-20134-5
Mar 3, 2026
Merged

KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (5/N)#21497
mjsax merged 12 commits intoapache:trunkfrom
frankvicky:KAFKA-20134-5

Conversation

@frankvicky
Copy link
Copy Markdown
Contributor

@frankvicky frankvicky commented Feb 17, 2026 •
edited by github-actions Bot
Loading

This PR adds required classes or modifies the existing ones to build the
TimestampedWindowStoreWithHeaders introduced in KIP-1271.

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

@github-actions github-actions Bot added triage PRs from the community streams labels Feb 17, 2026
@frankvicky frankvicky force-pushed the KAFKA-20134-5 branch 2 times, most recently from 7d824e8 to f9e684a Compare February 20, 2026 17:29
aliehsaeedii
aliehsaeedii reviewed Feb 23, 2026
View reviewed changes
Copy link
Copy Markdown
Contributor

@aliehsaeedii aliehsaeedii left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @frankvicky . I made a pass.

Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
public TimestampedToHeadersWindowStoreAdapter(final WindowStore<Bytes, byte[]> store) {
if (!store.persistent()) {
throw new IllegalArgumentException("Provided store must be a persistent store, but it is not.");
}
Copy link
Copy Markdown
Contributor

@aliehsaeedii aliehsaeedii Feb 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

also check if store is not of type timestamed, and then throw exception since we dont support upgrade from normal window store to ts-window-store-with-headers

Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java

@SuppressWarnings("unchecked")
@Override
public <R> QueryResult<R> query(final Query<R> query,
Copy link
Copy Markdown
Contributor

@aliehsaeedii aliehsaeedii Feb 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we should throw unsupportedoperationexception here as we dont support IQv2 yet.

Comment thread .../java/org/apache/kafka/streams/state/internals/TimestampedWindowStoreWithHeadersBuilder.java
}

@Override
public <R> QueryResult<R> query(final Query<R> query,
Copy link
Copy Markdown
Contributor

@aliehsaeedii aliehsaeedii Feb 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don' t support IQv2. Please just throw exception here.

@github-actions github-actions Bot removed the triage PRs from the community label Feb 24, 2026
mjsax
mjsax reviewed Feb 25, 2026
View reviewed changes
Comment thread streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java
Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java Outdated
Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java Outdated
@Override
public KeyValue<Windowed<Bytes>, byte[]> next() {
final KeyValue<Windowed<Bytes>, byte[]> timestampedKeyValue = innerIterator.next();
if (timestampedKeyValue == null) {
Copy link
Copy Markdown
Member

@mjsax mjsax Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This null check is missing in TimestampedToHeadersIteratorAdapter<K> -- not sure why.

Might be a bug we should address?

Copy link
Copy Markdown
Contributor Author

@frankvicky frankvicky Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we address in this PR or in another minor PR?

Copy link
Copy Markdown
Member

@mjsax mjsax Feb 27, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we address in this PR or in another minor PR?

Not sure. Whatever you prefer.

Comment thread ...in/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java Outdated
* 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[]> {
Copy link
Copy Markdown
Member

@mjsax mjsax Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems this one is very similar to KeyValueToTimestampedKeyValueIteratorAdapter just with the difference of the "converter function".

If we make the "converter function" a parameter, we would unify the code and just pass in different function.

Let's file a ticket to do this unification to reduce duplicated code. We should file this ticket not limited for this case, but look across all of the adaptor code, to see what we can unify. This might not be the only case of unnecessary code duplication.

Copy link
Copy Markdown
Contributor Author

@frankvicky frankvicky Feb 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://issues.apache.org/jira/browse/KAFKA-20224

@frankvicky frankvicky marked this pull request as ready for review February 25, 2026 12:52
@frankvicky frankvicky mentioned this pull request Feb 25, 2026
Merged
@frankvicky frankvicky mentioned this pull request Feb 27, 2026
Merged
@mjsax mjsax added the kip Requires or implements a KIP label Mar 3, 2026
@mjsax mjsax merged commit 7ed35ab into apache:trunk Mar 3, 2026
24 checks passed
frankvicky added a commit that referenced this pull request Mar 6, 2026
@frankvicky
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) (#21581)
5fc5d75 
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>
nicktelford pushed a commit to nicktelford/kafka that referenced this pull request Mar 6, 2026
@frankvicky @nicktelford
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (5/N) (apach…
041412c 
…e#21497)

This PR adds required classes or modifies the existing ones to build the
`TimestampedWindowStoreWithHeaders` introduced in KIP-1271.

Reviewers: Alieh Saeedi <asaeedi@confluent.io>, Matthias J. Sax
 <matthias@confluent.io>
gabriellefu pushed a commit to gabriellefu/kafka that referenced this pull request Mar 30, 2026
@frankvicky @gabriellefu
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) (apach…
30e8e64 
…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>
gabriellefu pushed a commit to gabriellefu/kafka that referenced this pull request Mar 30, 2026
@frankvicky @gabriellefu
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (5/N) (apach…
9c4351f 
…e#21497)

This PR adds required classes or modifies the existing ones to build the
`TimestampedWindowStoreWithHeaders` introduced in KIP-1271.

Reviewers: Alieh Saeedi <asaeedi@confluent.io>, Matthias J. Sax
 <matthias@confluent.io>
Shekharrajak pushed a commit to Shekharrajak/kafka that referenced this pull request Mar 31, 2026
@frankvicky @Shekharrajak
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (5/N) (apach…
14d0caa 
…e#21497)

This PR adds required classes or modifies the existing ones to build the
`TimestampedWindowStoreWithHeaders` introduced in KIP-1271.

Reviewers: Alieh Saeedi <asaeedi@confluent.io>, Matthias J. Sax
 <matthias@confluent.io>
Shekharrajak pushed a commit to Shekharrajak/kafka that referenced this pull request Mar 31, 2026
@frankvicky @Shekharrajak
KAFKA-20134: Implement TimestampedWindowStoreWithHeaders (N/N) (apach…
e0f4002 
…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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@mjsax mjsax mjsax approved these changes

+1 more reviewer

@aliehsaeedii aliehsaeedii aliehsaeedii left review comments

Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

kip Requires or implements a KIP streams

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@frankvicky @mjsax @aliehsaeedii