Jess668 opened a new pull request, #22799: URL: https://github.com/apache/kafka/pull/22799
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](https://github.com/apache/kafka/pull/22677) — already in `trunk`. 2. `ReadOnlyRecordIterator` [PR#22729](https://github.com/apache/kafka/pull/22729) — already in `trunk`. 3. `TimestampedKeyWithHeadersQuery` [PR#22666](https://github.com/apache/kafka/pull/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 `ReadOnlyRecord`s (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 - **Unit** — `MeteredTimestampedWindowStoreWithHeadersTest`: the typed query is translated into a raw `WindowKeyQuery` carrying the serialized key and the window-start bounds. - **Builder / end-to-end** — `TimestampedWindowStoreWithHeadersBuilderTest` (`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 store** — `RocksDBTimestampedWindowStoreWithHeadersTest` / `RocksDBTimeOrderedWindowStoreWithHeadersTest`: `WindowKeyQuery` is now handled (was `UNKNOWN_QUERY_TYPE`); `WindowRangeQuery` remains `UNKNOWN_QUERY_TYPE`. - **Integration** — `IQv2HeadersStoreIntegrationTest`: `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. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
