aliehsaeedii commented on code in PR #22882:
URL: https://github.com/apache/kafka/pull/22882#discussion_r3726275364
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -370,10 +370,198 @@ You can now find and query your custom store:
streams.start();
// Get access to the custom store
- MyReadableCustomStore<String,String> store =
streams.store("the-custom-store", new MyCustomStoreType<String,String>());
+ MyReadableCustomStore<String,String> store =
+ streams.store(StoreQueryParameters.fromNameAndType("the-custom-store",
new MyCustomStoreType<String,String>()));
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries
{#header-aware-stores-interactive-queries}
Review Comment:
[order] Placement: this section lands *after* `## Querying local custom
state stores` (line 260), which puts a built-in store type behind the
"implement your own" escape hatch.
The store-type list at lines 141-148 interleaves the three `*WithHeaders()`
helpers with the other built-ins, and each links here:
```
* QueryableStoreTypes#timestampedKeyValueStoreWithHeaders() ->
#header-aware-stores-interactive-queries
* QueryableStoreTypes#timestampedWindowStoreWithHeaders() ->
#header-aware-stores-interactive-queries
* QueryableStoreTypes#sessionStoreWithHeaders() ->
#header-aware-stores-interactive-queries
```
A reader walking the page top-to-bottom therefore gets: key-value -> window
-> *custom stores* -> header-aware. Line 149 ("You can also implement your own
QueryableStoreType…") reads as the closing note of the built-in helpers, so
custom stores is naturally the last section.
Suggest moving this whole `##` section to sit immediately before `##
Querying local custom state stores`, i.e. after `## Querying local window
stores`. That matches the list order and keeps custom stores at the end. It
would also help with the #22896 sequencing you agreed to — the IQv2 primer
moves ~120 lines earlier, closer to wherever that PR's IQv2 section ends up.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -370,10 +370,198 @@ You can now find and query your custom store:
streams.start();
// Get access to the custom store
- MyReadableCustomStore<String,String> store =
streams.store("the-custom-store", new MyCustomStoreType<String,String>());
+ MyReadableCustomStore<String,String> store =
+ streams.store(StoreQueryParameters.fromNameAndType("the-custom-store",
new MyCustomStoreType<String,String>()));
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries
{#header-aware-stores-interactive-queries}
+
+A [header-aware
store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores)
— built with a `*WithHeaders` supplier and its matching builder
([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists
each record's
[headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers())
together with its value and timestamp. This section shows how to read those
headers back interactively, through both the legacy `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes`
that matches your store type. The returned `ReadOnly*Store` surfaces the
headers as part of its value type:
[ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html)
for key-value and window stores, and
[AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html)
for session stores.
+
+
+ // Key-value store built with a *WithHeaders supplier
+ ReadOnlyKeyValueStore<String, ValueTimestampHeaders<Long>> keyValueStore =
+ streams.store(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());
+ }
+
+Session stores return `AggregationWithHeaders<V>`, which exposes the
aggregated value via `aggregation()` and the headers via `headers()`.
+
+### Reading headers with the IQv2 `query()` API
+
+Interactive Queries v2 (IQv2) is the query-based interactive-queries API:
instead of accessing a store object directly, you build a `Query`, wrap it in a
[StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html),
and run it with `KafkaStreams#query(...)`. The call returns a
[StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html)
that holds a per-partition
[QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html):
use `getOnlyPartitionResult()` for a single-key lookup, or
`getPartitionResults()` for the full `Map<Integer, QueryResult<R>>`. Each
`QueryResult` exposes the query result via `getResult()` and its data-freshness
`Position` via `getPosition()`.
+
+Before KIP-1356, no IQv2 query type exposed record headers. KIP-1356 adds four
`@Evolving` query types whose results carry headers. Each returns a
[ReadOnlyRecord](/{version}/javadoc/org/apache/kafka/streams/processor/api/ReadOnlyRecord.html)
— a read-only view exposing `key()`, `value()`, `timestamp()`, and `headers()`
— or, for the range and window queries, a closeable
[ReadOnlyRecordIterator](/{version}/javadoc/org/apache/kafka/streams/state/ReadOnlyRecordIterator.html)
of such records. `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>
+
+—
+</td>
+<td>
+
+Point query fails with a store-exception error; 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.)
Review Comment:
[prior comment] This is the one part of my earlier comment that didn't fully
land. The row states the plain-supplier point query as an unconditional
failure, but the failure is **cache-dependent**, and the `Headers` cell (`—`)
is wrong for the cache-served case.
With caching enabled and a warm entry, `TimestampedKeyWithHeadersQuery` on a
plain-adapter store **succeeds with the real value, timestamp, and headers** —
the value never went through the adapter, it came straight from the cache the
metered layer wrote. `STORE_EXCEPTION` only appears once the read is
store-served: after a flush, or with `skipCache()`.
The test asserts exactly this flip on a single instance
(`TimestampedKeyValueStoreBuilderWithHeadersTest.java:432-467`):
```java
if (cachingEnabled) {
assertTrue(result.isSuccess(), "Expected a cache-served read to
succeed");
...
assertEquals(123L, record.timestamp());
assertEquals(headers, record.headers());
// then flushCache() + skipCache() ->
assertEquals(FailureReason.STORE_EXCEPTION,
afterFlush.getFailureReason());
} else {
assertFalse(result.isSuccess(),
"A store-served read on a plain build has ts=-1 and must fail, not
return empty headers");
}
```
The mechanism is the `-1` timestamp injected on the way out of the store
(`HeadersBytesStore.java:66-90`), which
`MeteredTimestampedKeyValueStoreWithHeaders` turns into a failure rather than
letting `new Record<>` throw (`:422-432`) — so it can only bite a read that
actually reaches the store.
Suggested wording for the two cells:
- **Headers:** `Returned while cache-served; otherwise —`
- **Query outcome:** `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.)`
Note this is the same "store-served" qualifier you already applied correctly
to the timestamped row's prose — and line 563 (read-your-writes) is now
accurate, so the two statements just need to stop contradicting each other.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -184,7 +182,8 @@ After the application has started, you can get access to
"CountsKeyValueStore" a
// Get the key-value store CountsKeyValueStore
ReadOnlyKeyValueStore<String, Long> keyValueStore =
- streams.store("CountsKeyValueStore",
QueryableStoreTypes.keyValueStore());
+ streams.store(StoreQueryParameters.fromNameAndType(
Review Comment:
[content] This fix is right (and thanks for extending it to the pre-existing
examples), but it left the surrounding prose stale. **Line 138**, just above,
still says:
> Pass a built-in implementation from `QueryableStoreTypes` ... **as the
second argument to** `KafkaStreams#store(...)`.
There is no second argument any more — `store` takes exactly one parameter:
```java
public <T> T store(final StoreQueryParameters<T> storeQueryParameters) {
```
(`KafkaStreams.java:1867`, the only `store(` overload in the class; not
deprecated.)
After this PR the page's four examples all pass a single
`StoreQueryParameters`, so line 138 now contradicts every example below it.
Suggest rewording to something like "…pass a built-in implementation from
`QueryableStoreTypes` to `StoreQueryParameters.fromNameAndType(...)`, then hand
that to `KafkaStreams#store(...)`."
(Line 138 is outside the diff, hence the comment here.)
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -370,10 +370,198 @@ You can now find and query your custom store:
streams.start();
// Get access to the custom store
- MyReadableCustomStore<String,String> store =
streams.store("the-custom-store", new MyCustomStoreType<String,String>());
+ MyReadableCustomStore<String,String> store =
+ streams.store(StoreQueryParameters.fromNameAndType("the-custom-store",
new MyCustomStoreType<String,String>()));
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries
{#header-aware-stores-interactive-queries}
+
+A [header-aware
store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores)
— built with a `*WithHeaders` supplier and its matching builder
([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists
each record's
[headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers())
together with its value and timestamp. This section shows how to read those
headers back interactively, through both the legacy `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the legacy `store()` API
Review Comment:
[sentence] "legacy `store()` API" —
`KafkaStreams#store(StoreQueryParameters)` is not deprecated
(`KafkaStreams.java:1867` carries no `@Deprecated`; the deprecated-and-removed
thing was the old `store(String, QueryableStoreType)` overload). Calling it
"legacy" in a heading reads as "don't use this", which is the wrong signal —
for a header-aware store it's still the only way to get
`ValueTimestampHeaders`/`AggregationWithHeaders` off the store object, and it's
what the whole rest of this page teaches.
Suggest "Reading headers with the `store()` API" here, matching "Reading
headers with the IQv2 `query()` API" below, and dropping "legacy" from line 380
too ("…through both the `store()` API and the IQv2 `query()` API"). If you want
to mark the generational difference, "IQv1" is the term the KIPs use and it
carries no deprecation implication.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -370,10 +370,198 @@ You can now find and query your custom store:
streams.start();
// Get access to the custom store
- MyReadableCustomStore<String,String> store =
streams.store("the-custom-store", new MyCustomStoreType<String,String>());
+ MyReadableCustomStore<String,String> store =
+ streams.store(StoreQueryParameters.fromNameAndType("the-custom-store",
new MyCustomStoreType<String,String>()));
// Query the store
String value = store.read("key");
+## Header-aware stores and interactive queries
{#header-aware-stores-interactive-queries}
+
+A [header-aware
store](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores)
— built with a `*WithHeaders` supplier and its matching builder
([KIP-1271](../../upgrade-guide/#kip-1271-headers-aware-stores)) — persists
each record's
[headers](/{version}/javadoc/org/apache/kafka/streams/processor/api/Record.html#headers())
together with its value and timestamp. This section shows how to read those
headers back interactively, through both the legacy `store()` API and the IQv2
`query()` API.
+
+### Reading headers with the legacy `store()` API
+
+Look up the store with the `*WithHeaders()` entry from `QueryableStoreTypes`
that matches your store type. The returned `ReadOnly*Store` surfaces the
headers as part of its value type:
[ValueTimestampHeaders](/{version}/javadoc/org/apache/kafka/streams/state/ValueTimestampHeaders.html)
for key-value and window stores, and
[AggregationWithHeaders](/{version}/javadoc/org/apache/kafka/streams/state/AggregationWithHeaders.html)
for session stores.
Review Comment:
[completeness] This subsection only shows the key-value case, but three list
entries at lines 141-148 route readers here — including
`timestampedWindowStoreWithHeaders()` and `sessionStoreWithHeaders()`. Someone
who clicked the `sessionStoreWithHeaders()` bullet gets one sentence (line 399)
and no code.
Both helpers exist and their types are easy to state
(`QueryableStoreTypes.java:103` and `:125`):
```java
public static <K, V> QueryableStoreType<ReadOnlyWindowStore<K,
ValueTimestampHeaders<V>>> timestampedWindowStoreWithHeaders()
public static <K, V> QueryableStoreType<ReadOnlySessionStore<K,
AggregationWithHeaders<V>>> sessionStoreWithHeaders()
```
A two- or three-line snippet each (mirroring the key-value one) would make
all three inbound links pay off — the window one especially, since the
`ReadOnlyWindowStore<K, ValueTimestampHeaders<V>>` nesting is the kind of thing
readers copy rather than derive.
Second, smaller gap: the examples query a store named `counts-store` that
the page never creates. The other sections each open with the topology that
defines their store (e.g. line 159 introduces `CountsKeyValueStore`). Either
reuse an existing name or add one line pointing at [Headers in State
Stores](/{version}/streams/developer-guide/processor-api/#headers-in-state-stores)
for how to build one — the section links there for the *concept* (line 380)
but never for the *builder*.
--
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]
