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

[Jess668]

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 PR#22770 — 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

• 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.