You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2026/08/04 10:21:48 UTC
Re: [PR] KAFKA-20682: Document IQv2 headers-aware queries (KIP-1356) [kafka]
aliehsaeedii commented on code in PR #22882:
URL: https://github.com/apache/kafka/pull/22882#discussion_r3711364032
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
Review Comment:
[content] This example does not compile — there is no two-argument `store(String, QueryableStoreType)` method. `KafkaStreams` exposes exactly one `store` overload:
```java
public <T> T store(final StoreQueryParameters<T> storeQueryParameters) {
```
(`KafkaStreams.java:1867` — that is the only match for `grep -n "public .* store("` in the whole class). The `(String, QueryableStoreType)` form was deprecated in 2.5 and removed in 3.0.
Correct form:
```java
ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
streams.store(StoreQueryParameters.fromNameAndType(
"counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders()));
```
(`StoreQueryParameters.java:42`)
Heads-up: the pre-existing example at line 187 (`streams.store("CountsKeyValueStore", QueryableStoreTypes.keyValueStore())`) has the same defect. It's outside this diff so not your bug, but since you're adding a second instance, it would be a nice drive-by to fix both — otherwise the page teaches an API that no longer exists in two places.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
+
+`TimestampedKeyWithHeadersQuery` is a single-key lookup against a header-aware key-value store, parallel to `TimestampedKeyQuery`:
+
+
+ TimestampedKeyWithHeadersQuery<String, Long> query =
+ TimestampedKeyWithHeadersQuery.withKey("hello");
+
+ StateQueryRequest<ReadOnlyRecord<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecord<String, Long>> result = streams.query(request);
+ ReadOnlyRecord<String, Long> record = result.getOnlyPartitionResult().getResult();
Review Comment:
[content] `result.getOnlyPartitionResult().getResult()` can blow up two different ways, and the `if (record != null)` guard runs too late to help with either.
1. **NPE when the key isn't held locally.** `getOnlyPartitionResult()` returns `null` if no partition produced anything:
```java
return nonempty.isEmpty() ? null : nonempty.get(0);
```
(`StateQueryResult.java:80`). The `.getResult()` call then NPEs before the null check is reached — exactly the "key not found" case a reader will try first.
2. **`IllegalArgumentException` on a failed partition.** The filter behind `getOnlyPartitionResult()` keeps failures, not just values:
```java
.filter(r -> r.isFailure() || r.getResult() != null)
```
(`StateQueryResult.java:69-73`), and on a failed result `getResult()` throws:
```java
public R getResult() {
throw new IllegalArgumentException(
```
(`FailedQueryResult.java:84-85`). So a partition that came back `NOT_PRESENT` (say, migrated during a rebalance) makes this line throw rather than return `null`.
Suggested shape:
```java
QueryResult<ReadOnlyRecord<String, Long>> partitionResult = result.getOnlyPartitionResult();
if (partitionResult != null && partitionResult.isSuccess()) {
ReadOnlyRecord<String, Long> record = partitionResult.getResult();
if (record != null) {
System.out.println("value: " + record.value());
System.out.println("headers: " + record.headers());
}
}
```
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
+
+`TimestampedKeyWithHeadersQuery` is a single-key lookup against a header-aware key-value store, parallel to `TimestampedKeyQuery`:
+
+
+ TimestampedKeyWithHeadersQuery<String, Long> query =
+ TimestampedKeyWithHeadersQuery.withKey("hello");
+
+ StateQueryRequest<ReadOnlyRecord<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecord<String, Long>> result = streams.query(request);
+ ReadOnlyRecord<String, Long> record = result.getOnlyPartitionResult().getResult();
+ if (record != null) {
+ System.out.println("value: " + record.value());
+ System.out.println("headers: " + record.headers());
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly from the underlying store (only this single-key query offers `skipCache()`).
+
+`TimestampedRangeWithHeadersQuery` is a key-range scan, parallel to `TimestampedRangeQuery`. It returns a `ReadOnlyRecordIterator`, so close it when done (for example, with try-with-resources). A range can span several local partitions, so iterate `getPartitionResults()`:
+
+
+ TimestampedRangeWithHeadersQuery<String, Long> query =
+ TimestampedRangeWithHeadersQuery.<String, Long>withRange("a", "n");
+
+ StateQueryRequest<ReadOnlyRecordIterator<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<String, Long>> result = streams.query(request);
+ for (QueryResult<ReadOnlyRecordIterator<String, Long>> partition : result.getPartitionResults().values()) {
+ try (ReadOnlyRecordIterator<String, Long> iterator = partition.getResult()) {
Review Comment:
[content] Same problem in the range example: `partition.getResult()` is called with no `isFailure()` guard, and on a failed partition it throws `IllegalArgumentException` (`FailedQueryResult.java:84-85`) rather than yielding an empty iterator. Because the call sits in the try-with-resources *resource initializer*, the throw escapes the `try` entirely — no iterator is acquired and the loop aborts on the first failed partition, abandoning the partitions after it.
A range query is precisely the case where partial failure is expected (the surrounding prose even says "a range can span several local partitions"), so the example should show the guard:
```java
for (QueryResult<ReadOnlyRecordIterator<String, Long>> partition : result.getPartitionResults().values()) {
if (partition.isFailure()) {
System.out.println("failed: " + partition.getFailureReason() + " - " + partition.getFailureMessage());
continue;
}
try (ReadOnlyRecordIterator<String, Long> iterator = partition.getResult()) {
```
For reference, the IQv2 range example in the sibling PR #22896 does include this `isFailure()` check — worth matching so the two sections don't teach different levels of care.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
Review Comment:
[completeness] "The result is an immutable snapshot" is stronger than the contract, and the sentence omits the part that can actually corrupt state. The `ReadOnlyRecord#headers()` javadoc:
> The returned `Headers` should be treated as read-only. Records served as IQv2 results from a state store have their headers frozen via `RecordHeaders.setReadOnly()`, so adding or removing a header throws `IllegalStateException`. **Freezing is not a deep copy, however: the `value()` bytes of an existing header can still be mutated in place, so callers must not modify them.**
(`ReadOnlyRecord.java:54-62`)
The doc captures the `IllegalStateException` half but not the second half — which is the dangerous one, because it fails silently rather than throwing. Suggest something like: "…throws `IllegalStateException`. The freeze is shallow, though: the byte array behind an individual header value can still be mutated in place, so treat header values as read-only too."
Everything else in this paragraph checks out: `ReadOnlyRecord` exposes exactly `key()`, `value()`, `timestamp()`, `headers()` (`ReadOnlyRecord.java:42-63`), `headers()` is documented "Never null", `ReadOnlyRecordIterator extends Iterator<ReadOnlyRecord<K, V>>, Closeable` (`ReadOnlyRecordIterator.java:42`), and all four query types are `@Evolving @InterfaceAudience.Public`.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
+
+`TimestampedKeyWithHeadersQuery` is a single-key lookup against a header-aware key-value store, parallel to `TimestampedKeyQuery`:
+
+
+ TimestampedKeyWithHeadersQuery<String, Long> query =
+ TimestampedKeyWithHeadersQuery.withKey("hello");
+
+ StateQueryRequest<ReadOnlyRecord<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecord<String, Long>> result = streams.query(request);
+ ReadOnlyRecord<String, Long> record = result.getOnlyPartitionResult().getResult();
+ if (record != null) {
+ System.out.println("value: " + record.value());
+ System.out.println("headers: " + record.headers());
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly from the underlying store (only this single-key query offers `skipCache()`).
+
+`TimestampedRangeWithHeadersQuery` is a key-range scan, parallel to `TimestampedRangeQuery`. It returns a `ReadOnlyRecordIterator`, so close it when done (for example, with try-with-resources). A range can span several local partitions, so iterate `getPartitionResults()`:
+
+
+ TimestampedRangeWithHeadersQuery<String, Long> query =
+ TimestampedRangeWithHeadersQuery.<String, Long>withRange("a", "n");
+
+ StateQueryRequest<ReadOnlyRecordIterator<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<String, Long>> result = streams.query(request);
+ for (QueryResult<ReadOnlyRecordIterator<String, Long>> partition : result.getPartitionResults().values()) {
+ try (ReadOnlyRecordIterator<String, Long> iterator = partition.getResult()) {
+ while (iterator.hasNext()) {
+ ReadOnlyRecord<String, Long> record = iterator.next();
+ System.out.println(record.key() + " -> " + record.value() + " " + record.headers());
+ }
+ }
+ }
+
+Use `withLowerBound`, `withUpperBound`, or `withNoBounds` for open-ended or full scans. Results are unordered by default; call `withAscendingKeys()` or `withDescendingKeys()` to fix the order, which is defined over the serialized `byte[]` of the keys rather than their logical order.
+
+`TimestampedWindowKeyWithHeadersQuery` fetches all windows for a single key within a window-start range from a header-aware window store, parallel to `WindowKeyQuery`. Its results are keyed by `Windowed<K>` (the window lives in the key; `timestamp()` is the stored record event-time). Execute and consume the `ReadOnlyRecordIterator` exactly as for the range query above:
+
+
+ TimestampedWindowKeyWithHeadersQuery<String, Long> query =
+ TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange(
+ "hello", Instant.ofEpochMilli(0), Instant.now());
+ // Result element type: ReadOnlyRecord<Windowed<String>, Long>
+
+`TimestampedWindowRangeWithHeadersQuery` is parallel to `WindowRangeQuery` and has two forms. Use `withWindowStartRange(timeFrom, timeTo)` against a header-aware window store to fetch every key across a window-start range, or `withKey(key)` against a header-aware session store to fetch all sessions for a key (for session results, `timestamp()` is the session-window end). As with `WindowRangeQuery`, each store accepts only its corresponding form; submitting the wrong form fails with an unknown-query-type error.
+
+
+ // Window store: every key across a window-start range
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byWindow =
+ TimestampedWindowRangeWithHeadersQuery.withWindowStartRange(
+ Instant.ofEpochMilli(0), Instant.now());
+
+ // Session store: all sessions for one key
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byKey =
+ TimestampedWindowRangeWithHeadersQuery.withKey("hello");
+
+**Behavior notes**
+
+ * **Window start range is required.** As with the existing window queries, `TimestampedWindowKeyWithHeadersQuery` and the `withWindowStartRange` form of `TimestampedWindowRangeWithHeadersQuery` require a closed window-start range — both `timeFrom` and `timeTo` must be present, and both bounds are inclusive.
+ * **Close iterators exactly once.** The range and window queries return a `ReadOnlyRecordIterator`; close it when you are done — always, even if a `next()` call throws partway through — or the underlying store iterator (and the store's `num-open-iterators` metric) leaks. A try-with-resources block does this correctly. The iterator does not support `remove()`.
+ * **Headers depend on how the store was built.** With a `*WithHeaders` builder over a native (RocksDB) header supplier, the queries return the stored headers. With a `*WithHeaders` builder over a non-header supplier the underlying store cannot persist headers, so the outcome depends on the supplier: over a *timestamped* supplier, store-served reads succeed with an empty `headers()`; over a *plain* supplier that keeps no timestamp, the entry has no representable timestamp, so `TimestampedKeyWithHeadersQuery` fails with a store-exception error while the range and window queries return an iterator that throws a `StreamsException` when it reaches that entry. (With caching enabled, a not-yet-evicted write is still served from the record cache with its original headers — read-your-writes.) Against a plain, non-`WithHeaders` store, the new query types are unsupported and fail cleanly with an unknown-query-type error.
Review Comment:
[content] "the range **and window** queries return an iterator that throws a `StreamsException` when it reaches that entry" over-generalizes: it is not true for the session-store form. `TimestampedWindowRangeWithHeadersQuery`'s own javadoc calls this out explicitly for `withKey(...)`:
> Session aggregations carry no per-record event-time of their own, so `ReadOnlyRecord#timestamp()` is filled from the session window's (inclusive) end timestamp. That value is validated non-negative when the window is constructed, so **-- unlike the window-store form above -- this form can never throw while iterating**: a session whose stored value deserializes to null is still returned, with `ReadOnlyRecord#value()` `null` (the window-store form instead throws for such an entry, as it has no event-time to surface).
(`TimestampedWindowRangeWithHeadersQuery.java`, class javadoc)
So the throwing behaviour covers `TimestampedKeyWithHeadersQuery`, `TimestampedRangeWithHeadersQuery`, `TimestampedWindowKeyWithHeadersQuery`, and the `withWindowStartRange` form of `TimestampedWindowRangeWithHeadersQuery` — but the `withKey`/session form returns the entry with a `null` `value()` instead. Suggest naming the forms rather than saying "the range and window queries", since a reader relying on this will write a try/catch that can never fire (or omit a null check that can).
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
+
+`TimestampedKeyWithHeadersQuery` is a single-key lookup against a header-aware key-value store, parallel to `TimestampedKeyQuery`:
+
+
+ TimestampedKeyWithHeadersQuery<String, Long> query =
+ TimestampedKeyWithHeadersQuery.withKey("hello");
+
+ StateQueryRequest<ReadOnlyRecord<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecord<String, Long>> result = streams.query(request);
+ ReadOnlyRecord<String, Long> record = result.getOnlyPartitionResult().getResult();
+ if (record != null) {
+ System.out.println("value: " + record.value());
+ System.out.println("headers: " + record.headers());
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly from the underlying store (only this single-key query offers `skipCache()`).
+
+`TimestampedRangeWithHeadersQuery` is a key-range scan, parallel to `TimestampedRangeQuery`. It returns a `ReadOnlyRecordIterator`, so close it when done (for example, with try-with-resources). A range can span several local partitions, so iterate `getPartitionResults()`:
+
+
+ TimestampedRangeWithHeadersQuery<String, Long> query =
+ TimestampedRangeWithHeadersQuery.<String, Long>withRange("a", "n");
+
+ StateQueryRequest<ReadOnlyRecordIterator<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<String, Long>> result = streams.query(request);
+ for (QueryResult<ReadOnlyRecordIterator<String, Long>> partition : result.getPartitionResults().values()) {
+ try (ReadOnlyRecordIterator<String, Long> iterator = partition.getResult()) {
+ while (iterator.hasNext()) {
+ ReadOnlyRecord<String, Long> record = iterator.next();
+ System.out.println(record.key() + " -> " + record.value() + " " + record.headers());
+ }
+ }
+ }
+
+Use `withLowerBound`, `withUpperBound`, or `withNoBounds` for open-ended or full scans. Results are unordered by default; call `withAscendingKeys()` or `withDescendingKeys()` to fix the order, which is defined over the serialized `byte[]` of the keys rather than their logical order.
+
+`TimestampedWindowKeyWithHeadersQuery` fetches all windows for a single key within a window-start range from a header-aware window store, parallel to `WindowKeyQuery`. Its results are keyed by `Windowed<K>` (the window lives in the key; `timestamp()` is the stored record event-time). Execute and consume the `ReadOnlyRecordIterator` exactly as for the range query above:
+
+
+ TimestampedWindowKeyWithHeadersQuery<String, Long> query =
+ TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange(
+ "hello", Instant.ofEpochMilli(0), Instant.now());
+ // Result element type: ReadOnlyRecord<Windowed<String>, Long>
+
+`TimestampedWindowRangeWithHeadersQuery` is parallel to `WindowRangeQuery` and has two forms. Use `withWindowStartRange(timeFrom, timeTo)` against a header-aware window store to fetch every key across a window-start range, or `withKey(key)` against a header-aware session store to fetch all sessions for a key (for session results, `timestamp()` is the session-window end). As with `WindowRangeQuery`, each store accepts only its corresponding form; submitting the wrong form fails with an unknown-query-type error.
+
+
+ // Window store: every key across a window-start range
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byWindow =
+ TimestampedWindowRangeWithHeadersQuery.withWindowStartRange(
+ Instant.ofEpochMilli(0), Instant.now());
+
+ // Session store: all sessions for one key
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byKey =
+ TimestampedWindowRangeWithHeadersQuery.withKey("hello");
+
+**Behavior notes**
+
+ * **Window start range is required.** As with the existing window queries, `TimestampedWindowKeyWithHeadersQuery` and the `withWindowStartRange` form of `TimestampedWindowRangeWithHeadersQuery` require a closed window-start range — both `timeFrom` and `timeTo` must be present, and both bounds are inclusive.
+ * **Close iterators exactly once.** The range and window queries return a `ReadOnlyRecordIterator`; close it when you are done — always, even if a `next()` call throws partway through — or the underlying store iterator (and the store's `num-open-iterators` metric) leaks. A try-with-resources block does this correctly. The iterator does not support `remove()`.
+ * **Headers depend on how the store was built.** With a `*WithHeaders` builder over a native (RocksDB) header supplier, the queries return the stored headers. With a `*WithHeaders` builder over a non-header supplier the underlying store cannot persist headers, so the outcome depends on the supplier: over a *timestamped* supplier, store-served reads succeed with an empty `headers()`; over a *plain* supplier that keeps no timestamp, the entry has no representable timestamp, so `TimestampedKeyWithHeadersQuery` fails with a store-exception error while the range and window queries return an iterator that throws a `StreamsException` when it reaches that entry. (With caching enabled, a not-yet-evicted write is still served from the record cache with its original headers — read-your-writes.) Against a plain, non-`WithHeaders` store, the new query types are unsupported and fail cleanly with an unknown-query-type error.
Review Comment:
[content] I traced this whole bullet through the builders, adapters and their tests. The **core of it is correct** — the native/adapter split is real, a timestamped supplier really does yield empty `headers()`, a plain supplier really does produce `STORE_EXCEPTION` for the point query and a mid-iteration `StreamsException` for the iterator queries, and a non-`WithHeaders` store really does fail with `UNKNOWN_QUERY_TYPE`. But three of its scope qualifiers are wrong:
**1. "With a `*WithHeaders` builder over a non-header supplier the underlying store cannot persist headers" is too strong — it holds only for *persistent* suppliers.** The builder branches three ways, not two:
```java
if (!(store instanceof HeadersBytesStore)) {
if (store.persistent()) {
if (store instanceof TimestampedBytesStore) {
store = new TimestampedToHeadersStoreAdapter(store);
} else {
store = new PlainToHeadersStoreAdapter(store);
}
} else {
store = new InMemoryTimestampedKeyValueStoreWithHeadersMarker(store);
}
}
```
(`TimestampedKeyValueStoreBuilderWithHeaders.java:64-79`; `TimestampedWindowStoreWithHeadersBuilder.java` has the same shape.) The in-memory path is a *marker*, not a converting adapter — the value-with-headers bytes are stored verbatim, so headers round-trip normally. The tests group `IN_MEMORY` with `NATIVE`, not with the adapters (`TimestampedKeyValueStoreBuilderWithHeadersTest.java:329-336`, `:634-640`; `TimestampedWindowStoreWithHeadersBuilderTest.java:337-343`). Suggest saying "over a non-header **persistent** supplier".
**2. The caching parenthetical only holds for the single-key query.** `CachingKeyValueStore` registers a cache handler for `KeyQuery` alone; every other query type is forwarded straight to the wrapped store (`CachingKeyValueStore.java:84-91` and `:138-140`). So `TimestampedRangeWithHeadersQuery` and both window/session header queries **bypass the record cache entirely** — a cache-only write is invisible to them, and with a position bound they fail `NOT_UP_TO_BOUND` instead (`IQv2HeadersStoreIntegrationTest.java:351-373`, `:485-507`, `:589-610`). As written, the read-your-writes promise reads as applying to the whole bullet. It's also why the plain-supplier failure needs the same "store-served" qualifier you correctly applied to the timestamped branch: with caching on and a warm entry, `TimestampedKeyWithHeadersQuery` *succeeds* with real value, timestamp and headers, and only turns into `STORE_EXCEPTION` after a flush or with `skipCache()` (`TimestampedKeyValueStoreBuilderWithHe
adersTest.java:440-459`).
**3. The plain-vs-timestamped distinction doesn't exist for session stores.** `SessionStoreWithHeadersBuilder.build()` has a single path — `SessionToHeadersStoreAdapter` — with no plain/timestamped branch (`SessionStoreWithHeadersBuilder.java:63-66`), so a blanket "over a *plain* supplier" doesn't map onto the session store type at all.
A table (supplier kind × query type → outcome) might carry this more cleanly than one long sentence, given there are now three supplier paths and two query shapes.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
Review Comment:
[coordination] This PR and **#22896** ("KAFKA-16262: Add IQv2 to Kafka Streams documentation", same author, opened one day later) both add IQv2 documentation to *this same file*, and neither references the other:
- **Both introduce IQv2 independently.** The `### Reading headers with the IQv2 query() API` subsection below opens with its own IQv2 primer (build a `Query`, wrap it in a `StateQueryRequest`, run `KafkaStreams#query(...)`, read `getOnlyPartitionResult()` / `getPartitionResults()` / `getResult()` / `getPosition()`). #22896 adds a full `# Interactive Queries v2 (IQv2)` section covering exactly that ground in more depth.
- **The reading order would be backwards.** This section lands at ~line 375, as a `##` under `# Querying local state stores...`; #22896 appends its IQv2 section at ~line 502, *after* `# Querying remote state stores for the entire app`. Land both as-is and this headers-specific IQv2 material appears ~130 lines before IQv2 is introduced at all.
- **The overlap is editorial, not textual.** Both PRs report `MERGEABLE` because the hunks sit at different offsets, so both can merge cleanly and still leave the page with two IQv2 introductions in the wrong order. Nothing will warn you.
Both PRs are drafts, so there's room to sequence this. My suggestion (also left on #22896): land this one first — it's smaller, self-contained, and scoped to a single KIP — then have #22896 move its IQv2 section up to follow `# Querying local state stores for an app instance`, which places it *before* this section, and reduce the primer above to a link into it. The alternative is to fold both into one restructure of the page.
Separately, credit where due: relocating `#header-aware-stores-interactive-queries` from the bare `<a id=...>` at old line 155 onto this heading is a real improvement — the four existing inbound links (`upgrade-guide.md:114` and lines 142/145/147 of this file) now land on the actual section instead of a one-line note, and `{#anchor}` on a heading is already the established pattern here (`upgrade-guide.md:98`, line 126 of this file).
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -374,6 +372,100 @@ You can now find and query your custom store:
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries {#header-aware-stores-interactive-queries}
+
+A [header-aware store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores) — built with a `*WithHeaders` supplier and its matching builder ([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists each record's [headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers()) together with its value and timestamp. This section shows how to read those headers back interactively, through both the legacy `store()` API and the IQv2 `query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes` that matches your store type. The returned `ReadOnly*Store` surfaces the headers as part of its value type: [ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html) for key-value and window stores, and [AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html) for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store("counts-store", QueryableStoreTypes.timestampedKeyValueStoreWithHeaders());
+
+ ValueTimestampHeaders<Long> vth = keyValueStore.get("hello");
+ if (vth != null) {
+ System.out.println("value: " + vth.value());
+ System.out.println("timestamp: " + vth.timestamp());
+ System.out.println("headers: " + vth.headers());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API: instead of accessing a store object directly, you build a `Query`, wrap it in a [StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html), and run it with `KafkaStreams#query(...)`. The call returns a [StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html) that holds a per-partition [QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html): use `getOnlyPartitionResult()` for a single-key lookup, or `getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each `QueryResult` exposes the query result via `getResult()` and its data-freshness `Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four `@Evolving` query types whose results carry headers. Each returns a [ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html) — a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()` — or, for the range and window queries, a closeable [ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html) of such records. The result is an immutable snapshot: `headers()` is never null (an empty `Headers` when the record had none) and must not be modified — calling a mutating method on it (for example `add(...)`) throws `IllegalStateException`.
+
+`TimestampedKeyWithHeadersQuery` is a single-key lookup against a header-aware key-value store, parallel to `TimestampedKeyQuery`:
+
+
+ TimestampedKeyWithHeadersQuery<String, Long> query =
+ TimestampedKeyWithHeadersQuery.withKey("hello");
+
+ StateQueryRequest<ReadOnlyRecord<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecord<String, Long>> result = streams.query(request);
+ ReadOnlyRecord<String, Long> record = result.getOnlyPartitionResult().getResult();
+ if (record != null) {
+ System.out.println("value: " + record.value());
+ System.out.println("headers: " + record.headers());
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly from the underlying store (only this single-key query offers `skipCache()`).
+
+`TimestampedRangeWithHeadersQuery` is a key-range scan, parallel to `TimestampedRangeQuery`. It returns a `ReadOnlyRecordIterator`, so close it when done (for example, with try-with-resources). A range can span several local partitions, so iterate `getPartitionResults()`:
+
+
+ TimestampedRangeWithHeadersQuery<String, Long> query =
+ TimestampedRangeWithHeadersQuery.<String, Long>withRange("a", "n");
+
+ StateQueryRequest<ReadOnlyRecordIterator<String, Long>> request =
+ StateQueryRequest.inStore("counts-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<String, Long>> result = streams.query(request);
+ for (QueryResult<ReadOnlyRecordIterator<String, Long>> partition : result.getPartitionResults().values()) {
+ try (ReadOnlyRecordIterator<String, Long> iterator = partition.getResult()) {
+ while (iterator.hasNext()) {
+ ReadOnlyRecord<String, Long> record = iterator.next();
+ System.out.println(record.key() + " -> " + record.value() + " " + record.headers());
+ }
+ }
+ }
+
+Use `withLowerBound`, `withUpperBound`, or `withNoBounds` for open-ended or full scans. Results are unordered by default; call `withAscendingKeys()` or `withDescendingKeys()` to fix the order, which is defined over the serialized `byte[]` of the keys rather than their logical order.
+
+`TimestampedWindowKeyWithHeadersQuery` fetches all windows for a single key within a window-start range from a header-aware window store, parallel to `WindowKeyQuery`. Its results are keyed by `Windowed<K>` (the window lives in the key; `timestamp()` is the stored record event-time). Execute and consume the `ReadOnlyRecordIterator` exactly as for the range query above:
+
+
+ TimestampedWindowKeyWithHeadersQuery<String, Long> query =
+ TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange(
+ "hello", Instant.ofEpochMilli(0), Instant.now());
+ // Result element type: ReadOnlyRecord<Windowed<String>, Long>
+
+`TimestampedWindowRangeWithHeadersQuery` is parallel to `WindowRangeQuery` and has two forms. Use `withWindowStartRange(timeFrom, timeTo)` against a header-aware window store to fetch every key across a window-start range, or `withKey(key)` against a header-aware session store to fetch all sessions for a key (for session results, `timestamp()` is the session-window end). As with `WindowRangeQuery`, each store accepts only its corresponding form; submitting the wrong form fails with an unknown-query-type error.
+
+
+ // Window store: every key across a window-start range
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byWindow =
+ TimestampedWindowRangeWithHeadersQuery.withWindowStartRange(
+ Instant.ofEpochMilli(0), Instant.now());
+
+ // Session store: all sessions for one key
+ TimestampedWindowRangeWithHeadersQuery<String, Long> byKey =
+ TimestampedWindowRangeWithHeadersQuery.withKey("hello");
+
+**Behavior notes**
+
+ * **Window start range is required.** As with the existing window queries, `TimestampedWindowKeyWithHeadersQuery` and the `withWindowStartRange` form of `TimestampedWindowRangeWithHeadersQuery` require a closed window-start range — both `timeFrom` and `timeTo` must be present, and both bounds are inclusive.
+ * **Close iterators exactly once.** The range and window queries return a `ReadOnlyRecordIterator`; close it when you are done — always, even if a `next()` call throws partway through — or the underlying store iterator (and the store's `num-open-iterators` metric) leaks. A try-with-resources block does this correctly. The iterator does not support `remove()`.
+ * **Headers depend on how the store was built.** With a `*WithHeaders` builder over a native (RocksDB) header supplier, the queries return the stored headers. With a `*WithHeaders` builder over a non-header supplier the underlying store cannot persist headers, so the outcome depends on the supplier: over a *timestamped* supplier, store-served reads succeed with an empty `headers()`; over a *plain* supplier that keeps no timestamp, the entry has no representable timestamp, so `TimestampedKeyWithHeadersQuery` fails with a store-exception error while the range and window queries return an iterator that throws a `StreamsException` when it reaches that entry. (With caching enabled, a not-yet-evicted write is still served from the record cache with its original headers — read-your-writes.) Against a plain, non-`WithHeaders` store, the new query types are unsupported and fail cleanly with an unknown-query-type error.
+ * **Existing query types are unchanged.** The pre-existing IQv2 query types (`KeyQuery`, `TimestampedKeyQuery`, `RangeQuery`, `TimestampedRangeQuery`, `WindowKeyQuery`, `WindowRangeQuery`) also run against header-aware stores, returning header-stripped results, and now behave identically whether the header store was built on the native or the adapter path.
Review Comment:
[content] "now behave identically whether the header store was built on the native or the adapter path" needs a qualifier: what's actually true is native ≡ **timestamped** adapter. On the *plain* adapter two things differ:
1. **Timestamps are wrong, not merely absent.** `convertFromPlainToHeaderFormat` injects a hard-coded timestamp of `-1` (`HeadersBytesStore.java:71-80` — the format comment spells out `timestamp = -1 (8 bytes in big-endian)`), so `TimestampedKeyQuery` / `TimestampedRangeQuery` return a `ValueAndTimestamp` whose timestamp is `-1` rather than a real event-time.
2. **The window queries change result type.** `MeteredTimestampedWindowStoreWithHeaders#isUnderlyingStoreTimestamped()` returns `false` for the plain adapter:
```java
if (store instanceof PlainToHeadersWindowStoreAdapter) {
return false; // Plain store doesn't preserve timestamps
}
```
(`MeteredTimestampedWindowStoreWithHeaders.java:758-770`), which makes `WindowKeyQuery` / `WindowRangeQuery` yield plain `V` instead of `ValueAndTimestamp<V>`.
The test suite matches that reading: `PLAIN_ADAPTER` is deliberately excluded from every parity and `shouldHandle*` case, appearing only in the two negative tests (`TimestampedKeyValueStoreBuilderWithHeadersTest.java:432`, `:700`), while the parity tests cover `NATIVE`, `ADAPTER` and `IN_MEMORY` (`:292-327`, `:583-607`, `:786-818`).
The rest of the bullet is accurate — all six listed query types really are handled on header-aware stores and really do return header-stripped results (`MeteredTimestampedKeyValueStoreWithHeaders.java:95-120`, `MeteredTimestampedWindowStoreWithHeaders.java:154-165`, `MeteredSessionStoreWithHeaders.java:196-197`). Just narrow "the adapter path" to "the timestamped adapter path".
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org