aliehsaeedii commented on code in PR #22882:
URL: https://github.com/apache/kafka/pull/22882#discussion_r3757068653
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
Review Comment:
[content] "persists each record's headers together with its value and
timestamp" is not true for session stores, which this section goes on to cover.
`AggregationWithHeaders` has no timestamp at all — its only accessors are:
```java
public AGG aggregation() // AggregationWithHeaders.java:83
public Headers headers() // :87
```
and `TimestampedWindowRangeWithHeadersQuery`'s javadoc spells out why:
"Session aggregations carry no per-record event-time of their own, so
`ReadOnlyRecord#timestamp()` is filled from the session window's (inclusive)
end timestamp." Your own line 483 relies on that fact.
Since this is the sentence that defines the concept for the whole section,
suggest: "…persists each record's headers alongside its value (and, for
key-value and window stores, its timestamp)."
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the `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. These examples assume a header-aware store built with a
`*WithHeaders` supplier, as shown in [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores).
There are only three such helpers — `timestampedKeyValueStoreWithHeaders()`,
`timestampedWindowStoreWithHeaders()`, and `sessionStoreWithHeaders()`; there
is no `*WithHeaders()` helper for a plain (non-timestamped) key-value or window
store.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "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());
+ }
+
+ // Window store built with a *WithHeaders supplier
+ ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>> windowStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-window-store",
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+
+ // fetch returns a WindowStoreIterator whose values carry headers
+ try (WindowStoreIterator<ValueTimestampHeaders<Long>> it =
+ windowStore.fetch("hello", Instant.ofEpochMilli(0),
Instant.now())) {
+ while (it.hasNext()) {
+ ValueTimestampHeaders<Long> wv = it.next().value;
+ System.out.println("value: " + wv.value() + " headers: " +
wv.headers());
+ }
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` (not `value()`) and the headers via
`headers()`.
+
+
+ // Session store built with a *WithHeaders supplier
+ ReadOnlySessionStore<String, AggregationWithHeaders<Long>> sessionStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-session-store",
QueryableStoreTypes.sessionStoreWithHeaders()));
+
+ try (KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> it =
+ sessionStore.fetch("hello")) {
+ while (it.hasNext()) {
+ AggregationWithHeaders<Long> awh = it.next().value;
+ System.out.println("aggregation: " + awh.aggregation());
+ System.out.println("headers: " + awh.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](../../upgrade-guide/#kip-1356-iqv2-header-queries), no IQv2
query type exposed record headers.
[KIP-1356](../../upgrade-guide/#kip-1356-iqv2-header-queries) 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. `headers()` is never null (an empty `Headers` when the record
had none) and must be treated as read-only: records served as IQv2 results have
their headers frozen, so adding or removing a header (for example `add(...)`)
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.
+
+`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);
+ 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());
+ }
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly
from the underlying store (of the four header-aware queries, only this
single-key one 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.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()) {
+ if (partition.isFailure()) {
+ System.out.println("failed: " + partition.getFailureReason() + " - " +
partition.getFailureMessage());
+ continue;
+ }
+ 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. It parallels
`WindowKeyQuery`, but with a different result shape: `WindowKeyQuery` returns a
`WindowStoreIterator<V>` keyed by the window-start `long`, whereas this query
returns a `ReadOnlyRecordIterator<Windowed<K>, V>` whose records are keyed by
`Windowed<K>` (the window lives in the key; `timestamp()` is the stored record
event-time). Build and consume it as for the range query above, but note the
`Windowed<String>` in the request and result types:
+
+
+ TimestampedWindowKeyWithHeadersQuery<String, Long> query =
+ TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange(
+ "hello", Instant.ofEpochMilli(0), Instant.now());
+
+ StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>> request =
+ StateQueryRequest.inStore("counts-window-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<Windowed<String>, Long>> result =
streams.query(request);
+ // Iterate result.getPartitionResults() and close each
ReadOnlyRecordIterator, as in the range example.
+
+`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. Both forms are
`Query<ReadOnlyRecordIterator<Windowed<K>, V>>` — including the session
`withKey` form, whose records are keyed by the session's `Windowed<K>`.
+
+
+ // 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");
+
+ // Both forms build and consume identically; the result element type is
ReadOnlyRecord<Windowed<String>, Long>:
+ StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>> request =
+ StateQueryRequest.inStore("counts-window-store").withQuery(byWindow);
Review Comment:
[content] "Both forms build and consume identically" — but the snippet then
hardcodes the **window** store, and the store is precisely the thing that is
*not* identical. `byKey` is the session form; submitting it to
`"counts-window-store"` is exactly the mispairing the prose four lines up warns
about:
> As with `WindowRangeQuery`, each store accepts only its corresponding
form; submitting the wrong form fails with an unknown-query-type error.
Confirmed in the class javadoc: "Submitting the `withWindowStartRange` form
to a session store, or the `withKey` form to a window store, fails with
`FailureReason#UNKNOWN_QUERY_TYPE`"
(`TimestampedWindowRangeWithHeadersQuery.java`, class javadoc).
A reader who takes "identically" at face value and swaps `byWindow` for
`byKey` in the shown request gets `UNKNOWN_QUERY_TYPE`. Suggest showing both
requests, since the store names are already defined above:
```java
StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>>
windowRequest =
StateQueryRequest.inStore("counts-window-store").withQuery(byWindow);
StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>>
sessionRequest =
StateQueryRequest.inStore("counts-session-store").withQuery(byKey);
```
and narrowing the comment to "Both forms have the same result type; each
must target its own store type."
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the `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. These examples assume a header-aware store built with a
`*WithHeaders` supplier, as shown in [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores).
There are only three such helpers — `timestampedKeyValueStoreWithHeaders()`,
`timestampedWindowStoreWithHeaders()`, and `sessionStoreWithHeaders()`; there
is no `*WithHeaders()` helper for a plain (non-timestamped) key-value or window
store.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "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());
+ }
+
+ // Window store built with a *WithHeaders supplier
+ ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>> windowStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-window-store",
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+
+ // fetch returns a WindowStoreIterator whose values carry headers
+ try (WindowStoreIterator<ValueTimestampHeaders<Long>> it =
+ windowStore.fetch("hello", Instant.ofEpochMilli(0),
Instant.now())) {
+ while (it.hasNext()) {
+ ValueTimestampHeaders<Long> wv = it.next().value;
+ System.out.println("value: " + wv.value() + " headers: " +
wv.headers());
+ }
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` (not `value()`) and the headers via
`headers()`.
+
+
+ // Session store built with a *WithHeaders supplier
+ ReadOnlySessionStore<String, AggregationWithHeaders<Long>> sessionStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-session-store",
QueryableStoreTypes.sessionStoreWithHeaders()));
+
+ try (KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> it =
+ sessionStore.fetch("hello")) {
+ while (it.hasNext()) {
+ AggregationWithHeaders<Long> awh = it.next().value;
+ System.out.println("aggregation: " + awh.aggregation());
+ System.out.println("headers: " + awh.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](../../upgrade-guide/#kip-1356-iqv2-header-queries), no IQv2
query type exposed record headers.
[KIP-1356](../../upgrade-guide/#kip-1356-iqv2-header-queries) 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. `headers()` is never null (an empty `Headers` when the record
had none) and must be treated as read-only: records served as IQv2 results have
their headers frozen, so adding or removing a header (for example `add(...)`)
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.
+
+`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);
+ 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());
+ }
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly
from the underlying store (of the four header-aware queries, only this
single-key one offers `skipCache()`).
Review Comment:
[content] "Call `skipCache()` on the query" reads as an in-place mutation,
but the query types are immutable — `skipCache()` returns a **new** instance
and discards nothing into the receiver:
```java
public TimestampedKeyWithHeadersQuery<K, V> skipCache() {
return new TimestampedKeyWithHeadersQuery<>(key, true);
}
```
(`TimestampedKeyWithHeadersQuery.java:81-83`)
The example directly above binds the query to a variable and reuses it, so
the natural reading of this sentence produces a silent no-op:
```java
TimestampedKeyWithHeadersQuery<String, Long> query =
TimestampedKeyWithHeadersQuery.withKey("hello");
query.skipCache(); // return value dropped -- still cache-served
```
Suggest showing the chained form instead: "Chain `skipCache()` when building
the query — `TimestampedKeyWithHeadersQuery.withKey(\"hello\").skipCache()` —
to bypass the record cache and read directly from the underlying store (of the
four header-aware queries, only this single-key one offers `skipCache()`)."
The scoping claim in the parenthetical is correct as written.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the `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. These examples assume a header-aware store built with a
`*WithHeaders` supplier, as shown in [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores).
There are only three such helpers — `timestampedKeyValueStoreWithHeaders()`,
`timestampedWindowStoreWithHeaders()`, and `sessionStoreWithHeaders()`; there
is no `*WithHeaders()` helper for a plain (non-timestamped) key-value or window
store.
Review Comment:
[completeness] This subsection gives the reader no way to know that what
`store()` returns also depends on how the store was built. The table below
covers exactly that, but it sits under `### Reading headers with the IQv2
query() API` and is written in IQv2 vocabulary ("Query outcome",
"unknown-query-type", "store-exception"), so a reader who only needs the
`store()` API never sees it — and the two adapter paths degrade *silently*
here, unlike IQv2:
- **Timestamped adapter** — `headers()` comes back empty, same as the
table's timestamped row. The examples on this page print `vth.headers()` with
no hint it can be empty.
- **Plain adapter** — worse: `ValueTimestampHeaders` performs no timestamp
validation (`ValueTimestampHeaders.java:40-99` — the constructor and both
factories only null-check the value), so `vth.timestamp()` returns the injected
`-1` with no error. The negative-timestamp guard that turns this into a
`STORE_EXCEPTION` exists *only* on the IQv2 path
(`MeteredTimestampedKeyValueStoreWithHeaders.java:421-429`); `get(K)` at `:135`
goes straight through `getInternal` with no such check.
So the identical store that makes `TimestampedKeyWithHeadersQuery` fail
loudly makes `store().get()` return a plausible-looking `ValueTimestampHeaders`
with a bogus timestamp. Suggest one sentence here pointing forward, e.g.: "What
the store returns also depends on the supplier the `*WithHeaders` builder wraps
— see [the table below](#header-aware-stores-interactive-queries); note that on
the adapter paths the `store()` API degrades silently (empty headers, or a `-1`
timestamp) rather than failing."
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the `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. These examples assume a header-aware store built with a
`*WithHeaders` supplier, as shown in [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores).
There are only three such helpers — `timestampedKeyValueStoreWithHeaders()`,
`timestampedWindowStoreWithHeaders()`, and `sessionStoreWithHeaders()`; there
is no `*WithHeaders()` helper for a plain (non-timestamped) key-value or window
store.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "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());
+ }
+
+ // Window store built with a *WithHeaders supplier
+ ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>> windowStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-window-store",
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+
+ // fetch returns a WindowStoreIterator whose values carry headers
+ try (WindowStoreIterator<ValueTimestampHeaders<Long>> it =
+ windowStore.fetch("hello", Instant.ofEpochMilli(0),
Instant.now())) {
+ while (it.hasNext()) {
+ ValueTimestampHeaders<Long> wv = it.next().value;
+ System.out.println("value: " + wv.value() + " headers: " +
wv.headers());
+ }
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` (not `value()`) and the headers via
`headers()`.
+
+
+ // Session store built with a *WithHeaders supplier
+ ReadOnlySessionStore<String, AggregationWithHeaders<Long>> sessionStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-session-store",
QueryableStoreTypes.sessionStoreWithHeaders()));
+
+ try (KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> it =
+ sessionStore.fetch("hello")) {
+ while (it.hasNext()) {
+ AggregationWithHeaders<Long> awh = it.next().value;
+ System.out.println("aggregation: " + awh.aggregation());
+ System.out.println("headers: " + awh.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()`.
Review Comment:
[completeness] This paragraph introduces `Position` on the read side
(`getPosition()`) but never the write side, yet the third behavior bullet at
line 399 depends on it: "with a position bound, fails with a not-up-to-bound
error." Nothing on the page says how a reader would set one, so "position
bound" arrives as an undefined term at the point where it matters most.
It is a one-clause fix here, where `StateQueryRequest` is already being
described: mention `StateQueryRequest#withPositionBound(...)` alongside
`getPosition()`. That also makes the bullet's failure mode actionable — right
now a reader can observe `NOT_UP_TO_BOUND` but has no pointer to what produced
it.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -135,7 +135,7 @@ Every application instance can directly query any of its
local state stores.
The _name_ of a state store is defined when you create the store. You can
create the store explicitly by using the Processor API or implicitly by using
stateful operations in the DSL.
-The _type_ of a state store is defined by `QueryableStoreType`. Pass a
built-in implementation from
[`QueryableStoreTypes`](/{version}/javadoc/org/apache/kafka/streams/state/QueryableStoreTypes.html)
as the second argument to `KafkaStreams#store(...)`. The available built-in
helpers are:
+The _type_ of a state store is defined by `QueryableStoreType`. Pass a
built-in implementation from
[`QueryableStoreTypes`](/{version}/javadoc/org/apache/kafka/streams/state/QueryableStoreTypes.html)
to `StoreQueryParameters.fromNameAndType(...)`, then hand that to
`KafkaStreams#store(...)`. The available built-in helpers are:
Review Comment:
[completeness] This PR makes `StoreQueryParameters` load-bearing on this
page — it now appears in all four `store()` examples — but leaves it as a bare
code span, while `QueryableStoreTypes` in this same sentence gets a javadoc
link. Suggest linking it too:
`/{version}/javadoc/org/apache/kafka/streams/StoreQueryParameters.html` (note
the package is `org.apache.kafka.streams`, not `...streams.state`).
Separately — genuinely optional, and fair to defer as outside this KIP's
scope — `StoreQueryParameters` has two builder methods this page never mentions:
```java
public StoreQueryParameters<T> withPartition(final Integer partition) //
StoreQueryParameters.java:54
public StoreQueryParameters<T> enableStaleStores() // :63
```
`grep -niE 'stale|withPartition'` over this file returns **zero** hits, yet
the page's entire subject is querying local state stores — and
`enableStaleStores()` is precisely what a reader needs in order to query during
restore or a rebalance, which is one of the most common IQ questions. Worth a
sentence here, or a follow-up PR.
(Disclosure: the current wording of this line is what I suggested in an
earlier round, so the missing link is my omission, not yours.)
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,233 @@ After the application has started, you can get access to
"CountsWindowStore" and
System.out.println("Count of 'world' @ time " + windowTimestamp + " is "
+ next.value);
}
+## 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 `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the `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. These examples assume a header-aware store built with a
`*WithHeaders` supplier, as shown in [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores).
There are only three such helpers — `timestampedKeyValueStoreWithHeaders()`,
`timestampedWindowStoreWithHeaders()`, and `sessionStoreWithHeaders()`; there
is no `*WithHeaders()` helper for a plain (non-timestamped) key-value or window
store.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "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());
+ }
+
+ // Window store built with a *WithHeaders supplier
+ ReadOnlyWindowStore<String, ValueTimestampHeaders<Long>> windowStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-window-store",
QueryableStoreTypes.timestampedWindowStoreWithHeaders()));
+
+ // fetch returns a WindowStoreIterator whose values carry headers
+ try (WindowStoreIterator<ValueTimestampHeaders<Long>> it =
+ windowStore.fetch("hello", Instant.ofEpochMilli(0),
Instant.now())) {
+ while (it.hasNext()) {
+ ValueTimestampHeaders<Long> wv = it.next().value;
+ System.out.println("value: " + wv.value() + " headers: " +
wv.headers());
+ }
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` (not `value()`) and the headers via
`headers()`.
+
+
+ // Session store built with a *WithHeaders supplier
+ ReadOnlySessionStore<String, AggregationWithHeaders<Long>> sessionStore =
+ streams.store(StoreQueryParameters.fromNameAndType(
+ "counts-session-store",
QueryableStoreTypes.sessionStoreWithHeaders()));
+
+ try (KeyValueIterator<Windowed<String>, AggregationWithHeaders<Long>> it =
+ sessionStore.fetch("hello")) {
+ while (it.hasNext()) {
+ AggregationWithHeaders<Long> awh = it.next().value;
+ System.out.println("aggregation: " + awh.aggregation());
+ System.out.println("headers: " + awh.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](../../upgrade-guide/#kip-1356-iqv2-header-queries), no IQv2
query type exposed record headers.
[KIP-1356](../../upgrade-guide/#kip-1356-iqv2-header-queries) 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. `headers()` is never null (an empty `Headers` when the record
had none) and must be treated as read-only: records served as IQv2 results have
their headers frozen, so adding or removing a header (for example `add(...)`)
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.
+
+`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);
+ 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());
+ }
+ }
+
+Call `skipCache()` on the query to bypass the record cache and read directly
from the underlying store (of the four header-aware queries, only this
single-key one 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.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()) {
+ if (partition.isFailure()) {
+ System.out.println("failed: " + partition.getFailureReason() + " - " +
partition.getFailureMessage());
+ continue;
+ }
+ 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. It parallels
`WindowKeyQuery`, but with a different result shape: `WindowKeyQuery` returns a
`WindowStoreIterator<V>` keyed by the window-start `long`, whereas this query
returns a `ReadOnlyRecordIterator<Windowed<K>, V>` whose records are keyed by
`Windowed<K>` (the window lives in the key; `timestamp()` is the stored record
event-time). Build and consume it as for the range query above, but note the
`Windowed<String>` in the request and result types:
+
+
+ TimestampedWindowKeyWithHeadersQuery<String, Long> query =
+ TimestampedWindowKeyWithHeadersQuery.withKeyAndWindowStartRange(
+ "hello", Instant.ofEpochMilli(0), Instant.now());
+
+ StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>> request =
+ StateQueryRequest.inStore("counts-window-store").withQuery(query);
+
+ StateQueryResult<ReadOnlyRecordIterator<Windowed<String>, Long>> result =
streams.query(request);
+ // Iterate result.getPartitionResults() and close each
ReadOnlyRecordIterator, as in the range example.
+
+`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. Both forms are
`Query<ReadOnlyRecordIterator<Windowed<K>, V>>` — including the session
`withKey` form, whose records are keyed by the session's `Windowed<K>`.
+
+
+ // 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");
+
+ // Both forms build and consume identically; the result element type is
ReadOnlyRecord<Windowed<String>, Long>:
+ StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>> request =
+ StateQueryRequest.inStore("counts-window-store").withQuery(byWindow);
+ StateQueryResult<ReadOnlyRecordIterator<Windowed<String>, Long>> result =
streams.query(request);
+ // Iterate result.getPartitionResults() and close each
ReadOnlyRecordIterator, as in the range example.
+
+**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()`.
+ * **Read-your-writes applies only to the single-key query.** Only
`TimestampedKeyWithHeadersQuery` reads through the record cache, so it sees a
write that has not yet been flushed to the store — unless you call
`skipCache()`, or the entry has already been flushed. The range, window, and
session queries bypass the cache entirely, so a not-yet-flushed write is
invisible to them and, with a position bound, fails with a not-up-to-bound
error.
+
+**How the store was built determines what the queries return.** For key-value
and window stores, the outcome depends on the supplier the `*WithHeaders`
builder wraps:
+
+<table>
+<tr>
+<th>
+
+`*WithHeaders` store built over…
+</th>
+<th>
+
+Headers
+</th>
+<th>
+
+Query outcome
+</th> </tr>
+<tr>
+<td>
+
+Native (RocksDB) header supplier
+</td>
+<td>
+
+Returned
+</td>
+<td>
+
+All succeed
+</td> </tr>
+<tr>
+<td>
+
+In-memory non-header supplier
+</td>
+<td>
+
+Returned (a marker keeps the header-format bytes verbatim)
+</td>
+<td>
+
+All succeed
+</td> </tr>
+<tr>
+<td>
+
+Persistent *timestamped* non-header supplier
+</td>
+<td>
+
+Empty
+</td>
+<td>
+
+All succeed
+</td> </tr>
+<tr>
+<td>
+
+Persistent *plain* non-header supplier
+</td>
+<td>
+
+Returned while cache-served; otherwise —
+</td>
+<td>
+
+Store-served point query fails with a store-exception error (a cache-served
read still succeeds, with real value, timestamp, and headers, until the cache
is flushed or `skipCache()` is used); the range, window-key, and
`withWindowStartRange` window-range iterators throw a `StreamsException`
mid-iteration. (The `withKey` form of the window-range query targets session
stores, covered in the note below.)
+</td> </tr>
+<tr>
+<td>
+
+*(no `*WithHeaders` builder at all)*
+</td>
+<td>
+
+—
+</td>
+<td>
+
+Unknown-query-type
+</td> </tr> </table>
+
+Session stores have no plain/timestamped split: a `*WithHeaders` session store
built over a non-header supplier uses a single adapter and behaves like the
*timestamped* row above. The `withKey` form of
`TimestampedWindowRangeWithHeadersQuery` (the session-store form) never throws
— a session window always carries a valid end timestamp — so it returns empty
`headers()` and surfaces a `null` `value()` only where the stored value itself
is null.
Review Comment:
[content] "uses a single adapter and behaves like the *timestamped* row
above" holds only for a **persistent** non-header supplier. The session builder
branches the same two ways the key-value/window builders do:
```java
if (!(sessionStore instanceof HeadersBytesStore)) {
if (sessionStore.persistent()) {
sessionStore = new SessionToHeadersStoreAdapter(sessionStore);
} else {
sessionStore = new
InMemorySessionStoreWithHeadersMarker(sessionStore);
}
}
```
(`SessionStoreWithHeadersBuilder.java:62-69`)
`InMemorySessionStoreWithHeadersMarker` (`:97-99`) is a marker — it
`implements HeadersBytesStore` and wraps without converting, so the
header-format bytes round-trip verbatim. An **in-memory** session store built
over a non-header supplier therefore behaves like the *in-memory* row of the
table (headers returned), not the *timestamped* row (headers empty).
This is the same distinction you correctly added to the table for key-value
and window stores; the session footnote just didn't get it. Suggest: "…a
`*WithHeaders` session store built over a non-header **persistent** supplier
uses a single adapter and behaves like the *timestamped* row above; over an
in-memory supplier it uses a marker and behaves like the *in-memory* row."
The rest of the sentence — no plain/timestamped split, and the `withKey`
form never throwing — is correct.
--
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]
