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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
