Skip to content

KAFKA-20705: IQv2 TimestampedWindowKeyWithHeadersQuery for headers-aware window stores (KIP-1356)#22799

Draft
Jess668 wants to merge 17 commits into
apache:trunkfrom
Jess668:kip-1356-timestamped-window-key-with-headers-query
Draft

KAFKA-20705: IQv2 TimestampedWindowKeyWithHeadersQuery for headers-aware window stores (KIP-1356)#22799
Jess668 wants to merge 17 commits into
apache:trunkfrom
Jess668:kip-1356-timestamped-window-key-with-headers-query

Conversation

@Jess668

@Jess668 Jess668 commented Jul 9, 2026

Copy link
Copy Markdown
Contributor

Implements TimestampedWindowKeyWithHeadersQuery, the third IQv2 query type of KIP-1356 — the headers-aware parallel of WindowKeyQuery. Its result is an iterator of windowed records that each carry the windowed key, value, timestamp (record event-time), and headers. The result type is ReadOnlyRecordIterator<Windowed<K>, V> (already in trunk) — a Closeable iterator that yields ReadOnlyRecord<Windowed<K>, V> elements directly.

Dependencies

This work is stacked on, and should be reviewed/merged after:

  1. ReadOnlyRecord PR#22677 — already in trunk.
  2. ReadOnlyRecordIterator PR#22729 — already in trunk.
  3. TimestampedKeyWithHeadersQuery PR#22666 — already in trunk.
  4. TimestampedRangeWithHeadersQuery — the preceding KIP-1356 query type; this branch is stacked on it and should be rebased onto trunk once it lands, leaving only the window-key commits.

Summary

  1. TimestampedWindowKeyWithHeadersQuery<K, V> implements Query<ReadOnlyRecordIterator<Windowed<K>, V>>
    (@Evolving / @InterfaceAudience.Public). Parallels WindowKeyQuery — single static factory
    withKeyAndWindowStartRange(key, timeFrom, timeTo), a closed window-start range is required — but uses
    the newer no-get accessors (key(), timeFrom(), timeTo()) and rejects a null key, consistent
    with TimestampedKeyWithHeadersQuery / TimestampedRangeWithHeadersQuery. Each element's key is a
    Windowed<K> (the queried key + the entry's window); timestamp() is the stored record event-time.
  2. MeteredTimestampedWindowStoreWithHeaders handles the new query type by forwarding a raw
    WindowKeyQuery to the wrapped store and wrapping the returned WindowStoreIterator in a
    ReadOnlyRecordIterator whose elements are immutable ReadOnlyRecords (headers frozen via
    setReadOnly()). An entry whose stored timestamp is negative cannot be represented as a
    ReadOnlyRecord, so the lazily-evaluated iterator throws StreamsException while advancing.
  3. Native store enablement: RocksDBTimestampedWindowStoreWithHeaders and
    RocksDBTimeOrderedWindowStoreWithHeaders special-case WindowKeyQuery to the inherited store
    handling (StoreQueryUtils) via super.query(); previously they returned UNKNOWN_QUERY_TYPE for
    every query. Every other query type — including WindowRangeQuery — remains UNKNOWN_QUERY_TYPE and
    is deferred to a follow-up, mirroring how the TimestampedKeyWithHeadersQuery PR special-cased
    KeyQuery (rather than dropping the override outright).
  4. WindowKeyQuery constructor fix: its private constructor declared its parameters as
    (key, timeTo, timeFrom) while the withKeyAndWindowStartRange factory passes
    (key, timeFrom, timeTo), so getTimeFrom() / getTimeTo() returned each other's values. This was
    masked in the existing MeteredWindowStore path (which re-wraps the query through the same factory,
    cancelling the swap) but is exposed by the new handler and the now-enabled native path, both of which
    read the accessors once. Fixed the parameter order; verified net-invariant for the existing path
    (StoreQueryUtilsTest, MeteredWindowStoreTest, IQv2StoreIntegrationTest).

Testing

  • UnitMeteredTimestampedWindowStoreWithHeadersTest: the typed query is translated into a raw
    WindowKeyQuery carrying the serialized key and the window-start bounds.
  • Builder / end-to-endTimestampedWindowStoreWithHeadersBuilderTest (QueryTests):
    TimestampedWindowKeyWithHeadersQuery returns exact headers on a header-persisting (native) store,
    empty headers on the adapter store, throws on a plain supplier and on a negative stored timestamp;
    execution-info collect/no-collect; identical results across native- and adapter-built stores.
  • Native storeRocksDBTimestampedWindowStoreWithHeadersTest /
    RocksDBTimeOrderedWindowStoreWithHeadersTest: WindowKeyQuery is now handled (was
    UNKNOWN_QUERY_TYPE); WindowRangeQuery remains UNKNOWN_QUERY_TYPE.
  • IntegrationIQv2HeadersStoreIntegrationTest: TimestampedWindowKeyWithHeadersQuery over a
    running topology returns each record's windowed key/value/timestamp/headers across a window-start
    range (empty headers and read-only headers included, tombstoned window omitted); UNKNOWN_QUERY_TYPE
    against a non-headers store; failure (throws while iterating) against a plain supplier.

Jess668 added 17 commits July 8, 2026 10:45
- Add the headers-aware range query TimestampedRangeWithHeadersQuery, returning a
ReadOnlyRecordIterator of ReadOnlyRecord (key/value/timestamp/headers).
- MeteredTimestampedKeyValueStoreWithHeaders handles it by forwarding a raw RangeQuery
to the wrapped store and adapting each entry into a frozen-header ReadOnlyRecord;
an entry whose stored timestamp is negative cannot be represented as a ReadOnlyRecord,
so advancing the iterator throws StreamsException.
- Also drop the query() override on RocksDBTimestampedStoreWithHeaders so the native
header store serves the basic IQv2 queries via the inherited RocksDBStore handling.
- TimestampedKeyValueStoreBuilderWithHeadersTest: end-to-end build-path coverage
  (native/in-memory return headers; adapter returns empty headers; plain-legacy and
  negative timestamps throw while iterating), execution-info collection, and native/adapter parity.
- MeteredTimestampedKeyValueStoreWithHeadersTest: unit coverage of the metered handler
  forwarding the bounds and result order onto the raw RangeQuery.
- RocksDBTimestampedStoreWithHeadersTest: the native store now serves RangeQuery.
- TimestampedRangeWithHeadersQueryIntegrationTest: IQv2 integration coverage.
…ntion

- Query with PositionBound.at(INPUT_POSITION) + iqv2WaitForResult
- Introduce the headers-aware windowed point query, returning a ReadOnlyRecordIterator<Windowed<K>, V> whose elements carry the stored record headers.
- MeteredTimestampedWindowStoreWithHeaders handles it by forwarding a raw WindowKeyQuery to the wrapped store and wrapping each entry as a ReadOnlyRecord (Windowed key + value + timestamp + frozen headers).
- The native header window stores (RocksDBTimestampedWindowStoreWithHeaders, RocksDBTimeOrderedWindowStoreWithHeaders) special-case WindowKeyQuery to super.query() so it is served via StoreQueryUtils; every other query type (incl. WindowRangeQuery) stays UNKNOWN_QUERY_TYPE, deferred to a follow-up, mirroring the KV KeyQuery PR.
- Also fixes a latent WindowKeyQuery constructor bug that swapped timeFrom/timeTo.
@github-actions github-actions Bot added triage PRs from the community streams labels Jul 9, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

streams triage PRs from the community

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant