aliehsaeedii commented on code in PR #22882:
URL: https://github.com/apache/kafka/pull/22882#discussion_r3731111085
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
+
+### 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. `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 (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()) {
+ 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, 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:
Review Comment:
[content] Two problems here, both from the window queries having a different
result shape than the range query this snippet defers to.
**1. "parallel to `WindowKeyQuery`" hides a material divergence.** The two
resolve to different `Query` types:
```java
public final class TimestampedWindowKeyWithHeadersQuery<K, V> implements
Query<ReadOnlyRecordIterator<Windowed<K>, V>>
```
(`TimestampedWindowKeyWithHeadersQuery.java:71`)
```java
public class WindowKeyQuery<K, V> implements Query<WindowStoreIterator<V>>
```
(`WindowKeyQuery.java:29`)
`WindowKeyQuery` yields elements keyed by the window-start `long`; this one
yields records keyed by `Windowed<K>`. The prose does state the `Windowed<K>`
fact, but "parallel to" invites the reader to assume the same result shape, and
the difference is exactly the thing that breaks their code.
**2. "Execute and consume the `ReadOnlyRecordIterator` exactly as for the
range query above" does not compile if followed literally.** A reader copies
line 327:
```java
StateQueryRequest<ReadOnlyRecordIterator<String, Long>> request = // range
query
```
but the window key query needs:
```java
StateQueryRequest<ReadOnlyRecordIterator<Windowed<String>, Long>> request =
StateQueryRequest.inStore("counts-window-store").withQuery(query);
StateQueryResult<ReadOnlyRecordIterator<Windowed<String>, Long>> result =
streams.query(request);
```
The `// Result element type: ReadOnlyRecord<Windowed<String>, Long>` comment
is correct but easy to miss, and it names the *element* type while the compile
error is about the *request/result* type. Suggest spelling out the two
declarations instead of the comment.
The same gap applies to `TimestampedWindowRangeWithHeadersQuery` at lines
357-364 — **both** forms (including the session `withKey` form) are
`Query<ReadOnlyRecordIterator<Windowed<K>, V>>`
(`TimestampedWindowRangeWithHeadersQuery.java:85`), which is genuinely
surprising for the session case, and that snippet also stops before the request.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
Review Comment:
[completeness] Thanks for adding the window and session snippets — but
unlike the key-value one, both stop at the store lookup and never show a read,
which drops the one detail a reader cannot guess.
The session value accessor is **not** `value()`:
```java
public AGG aggregation()
```
(`AggregationWithHeaders.java:83`, with `headers()` at `:87`)
Line 286 says this in prose, but the accessor name is precisely the kind of
thing readers copy from code rather than parse from a sentence — and
`aggregation()` is unusual enough that guessing `value()` is the default
mistake. Suggest two more lines each, mirroring the key-value example:
```java
// Window store
WindowStoreIterator<ValueTimestampHeaders<Long>> it =
windowStore.fetch("hello", timeFrom, timeTo);
// Session store
AggregationWithHeaders<Long> awh = ...;
System.out.println("aggregation: " + awh.aggregation());
System.out.println("headers: " + awh.headers());
```
(Store type confirmed: `sessionStoreWithHeaders()` returns
`QueryableStoreType<ReadOnlySessionStore<K, AggregationWithHeaders<V>>>`,
`QueryableStoreTypes.java:125`.)
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
+
+### 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. `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.
Review Comment:
[completeness] KIP-1356 is named twice in this paragraph with no link, while
KIP-1271 four lines earlier gets one (line 262). More importantly, **KIP-1356
is undocumented everywhere else** — `grep -rn "1356" docs/streams/` returns
zero hits outside this file.
That's a gap for a change that adds six public types (four `@Evolving` query
classes plus `ReadOnlyRecord` and `ReadOnlyRecordIterator`):
- KIP-1271 has `upgrade-guide.md:98` (`### Header-aware state stores for the
Processor API (KIP-1271) {#kip-1271-headers-aware-stores}`) and KIP-1285 has
`:118` — so the neighbouring KIPs in this feature family each got an entry.
- `upgrade-guide.md:114` already promises "`TopologyTestDriver` and
Interactive Queries support the new store types" and links *here*, but says
nothing about the new IQv2 query types.
Two suggestions: (a) hyperlink both KIP-1356 mentions, matching the cwiki
style used throughout `upgrade-guide.md` (`:31, :60, :70, :72, :74, :76, :78`);
and (b) add a short KIP-1356 entry to `upgrade-guide.md` pointing at this
section — otherwise the only way to discover these query types is to already be
reading this page.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
+
+### 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. `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 (only this single-key query offers `skipCache()`).
Review Comment:
[sentence] "(only this single-key query offers `skipCache()`)" is false as
scoped. `skipCache()` also exists on the two pre-existing point queries:
- `KeyQuery.java:57`
- `TimestampedKeyQuery.java:59`
- `TimestampedKeyWithHeadersQuery.java:81`
The intended claim — true — is that of the *four header-aware* queries, only
this one has it (`TimestampedRangeWithHeadersQuery`,
`TimestampedWindowKeyWithHeadersQuery`, and
`TimestampedWindowRangeWithHeadersQuery` have no `skipCache()`/`isSkipCache()`
at all). Suggest "of the four header-aware queries, only this single-key one
offers `skipCache()`".
Optional but genuinely useful: `skipCache` is currently propagated to the
cache layer only by the *header* point query
(`MeteredTimestampedKeyValueStoreWithHeaders.java:413-417`, honored at
`CachingKeyValueStore.java:181-183`); that code comments that plain
`KeyQuery`/`TimestampedKeyQuery` do **not** yet propagate it (KAFKA-20776). So
on a header-aware store `skipCache()` actually works end-to-end here in a way
it doesn't for the older queries — worth a clause if you want the parenthetical
to earn its space.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
+
+### 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. `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 (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()) {
+ 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, 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()`.
+ * **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 *timestamped* adapter path. (The
*plain* adapter is not equivalent: it surfaces a `-1` timestamp rather than a
real event-time, and its window queries return plain values instead of
`ValueAndTimestamp`.)
Review Comment:
[order] This bullet is a forward reference: it turns on the distinction
between "the native", "the *timestamped* adapter path", and "the *plain*
adapter" — but nothing above defines those three paths. They are introduced by
the table immediately **below** it (lines 372-452: "Native (RocksDB) header
supplier", "Persistent *timestamped* non-header supplier", "Persistent *plain*
non-header supplier").
So a reader hits "now behave identically whether the header store was built
on the native or the *timestamped* adapter path" with no idea that a store has
adapter paths at all — the section intro (line 262) only mentions "a
`*WithHeaders` supplier".
Simplest fix: move this third bullet to just **after** the table, where the
vocabulary exists. The other two bullets (window-start range, closing
iterators) have no such dependency and can stay where they are.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
Review Comment:
[completeness] "Look up the store with the `*WithHeaders()` entry from
`QueryableStoreTypes` that matches your store type" implies an entry exists for
every store type. There are only three, and none covers a *plain*
(non-timestamped) key-value or window store:
```java
timestampedKeyValueStoreWithHeaders() // QueryableStoreTypes.java:70
timestampedWindowStoreWithHeaders() // :103
sessionStoreWithHeaders() // :125
```
Those are the only `*WithHeaders` factories in the class (8 public factories
total, at `:48 :59 :70 :81 :92 :103 :114 :125`) — there is no
`keyValueStoreWithHeaders()` and no `windowStoreWithHeaders()`. A reader with a
plain store will go looking for one.
The sibling page already states the reason, so this can be a single borrowed
clause:
> There are no `WithHeaders` suppliers for plain persistent key-value or
plain persistent windowed stores. `WithHeaders` suppliers exist only for
persistent timestamped key-value, persistent timestamped windowed, and session
stores.
(`processor-api.md:182`)
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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:
[sentence] Minor markup consistency: the `headers` link leaves the `()` bare
in the destination —
```
[headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers())
```
The parens are balanced so goldmark should handle it, but every other
javadoc link with a method signature in these docs wraps the destination in
angle brackets, including the two that point at this exact anchor:
- `processor-api.md:182` — `[record
headers](<https://kafka.apache.org/{version}/javadoc/.../Record.html#headers()>)`
- `processor-api.md:309` — same form
- `processor-api.md:176-180` —
`Stores.html#persistentKeyValueStore(java.lang.String)` etc., all angle-wrapped
This is the only bare-paren instance in `docs/streams/`. Suggest
`](<...#headers()>)` to match — cheap insurance against a renderer that is
stricter about parens.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -257,6 +257,204 @@ 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).
+
+
+ // 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()));
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` 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()));
+
+### 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. `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 (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()) {
+ 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, 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()`.
+ * **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 *timestamped* adapter path. (The
*plain* adapter is not equivalent: it surfaces a `-1` timestamp rather than a
real event-time, and its window queries return plain values instead of
`ValueAndTimestamp`.)
+
+**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.
+
+Read-your-writes applies only to the single-key
`TimestampedKeyWithHeadersQuery`, which reads through the record cache; the
range, window, and session queries bypass the cache, so a not-yet-flushed write
is invisible to them and, with a position bound, fails with a not-up-to-bound
error.
Review Comment:
[order] This read-your-writes paragraph is accurate — I confirmed the whole
thing (`CachingKeyValueStore.java:85-91` registers a handler for `KeyQuery`
alone and `:138-141` forwards every other query type down;
`CachingWindowStore`/`CachingSessionStore` have no `query()` override at all,
so `WrappedStateStore.query()` passes through; `NOT_UP_TO_BOUND` comes from the
store's own position check at `StoreQueryUtils.java:134-139`).
The problem is placement. Caching behaviour is the most consequential
runtime fact in this section for anyone building an RPC layer on top of IQv2,
and it currently sits as an unheaded trailing paragraph *after* a 5-row table
about store construction and a session-store footnote — i.e. the reader most
likely to skim the table is the one who misses it.
It is also a **behaviour** note, and there is a "Behavior notes" list at
lines 366-370 that is the natural home. Suggest promoting it to a fourth bullet
there (`* **Read-your-writes applies only to the single-key query.**`). One
tradeoff worth handling: that moves it ~68 lines *above* the plain-row caching
parenthetical at line 438 that it complements, so the bullet should spell out
the `skipCache()`/post-flush condition itself rather than leave the reader to
connect it to that cell.
--
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]
