aliehsaeedii commented on code in PR #22896:
URL: https://github.com/apache/kafka/pull/22896#discussion_r3711340098
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -499,6 +499,198 @@ At this point the full state of the application is
interactively queryable:
+# Interactive Queries v2 (IQv2) {#interactive-queries-v2}
+
+The sections above describe the original interactive queries API (informally,
"IQv1"), where you obtain a read-only store facade from
[KafkaStreams#store(...)](/{version}/javadoc/org/apache/kafka/streams/KafkaStreams.html)
and call methods such as `get(key)` or `range(from, to)` on it. Interactive
Queries v2 (IQv2), introduced in
[KIP-796](https://cwiki.apache.org/confluence/x/34xnCw), is an alternative,
more flexible API for the same purpose: querying the local state of a running
Kafka Streams application.
+
+Instead of a fixed store facade, IQv2 models every query as a first-class
[Query](/{version}/javadoc/org/apache/kafka/streams/query/Query.html) object
that you submit through a single
[KafkaStreams#query(StateQueryRequest)](/{version}/javadoc/org/apache/kafka/streams/KafkaStreams.html)
method. The response is a
[StateQueryResult](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryResult.html)
that contains a separate
[QueryResult](/{version}/javadoc/org/apache/kafka/streams/query/QueryResult.html)
for each partition that executed the query, along with metadata such as each
partition's
[Position](/{version}/javadoc/org/apache/kafka/streams/query/Position.html).
+
+IQv2 offers several advantages over the original API:
+
+ * A single, uniform entry point for every kind of query.
+ * Query-level extensibility: custom state stores can handle their own
`Query` types (the runtime forwards unknown queries straight through to the
underlying byte store), rather than requiring a custom `QueryableStoreType`
store facade as the original API does.
+ * Rich, per-partition results with typed failure reasons instead of
exceptions.
+ * Fine-grained consistency control through `Position` and `PositionBound`.
+ * The ability to target specific partitions, require active (leader) tasks,
override the isolation level, and collect execution information.
+
+Both APIs are fully supported and can be used side by side. Note that IQv2 is
marked as an evolving API, so it may change between releases, and global stores
are not yet supported by `query(...)` (see [Limitations](#limitations-of-iqv2)
below).
+
+## How Interactive Queries v2 works
+
+Issuing an IQv2 query follows four steps:
+
+ 1. **Build a query.** Create a `Query` object that describes what you want
to read, for example a
[KeyQuery](/{version}/javadoc/org/apache/kafka/streams/query/KeyQuery.html) for
a single-key lookup or a
[RangeQuery](/{version}/javadoc/org/apache/kafka/streams/query/RangeQuery.html)
for a scan.
+ 2. **Build a request.** Wrap the query in a
[StateQueryRequest](/{version}/javadoc/org/apache/kafka/streams/query/StateQueryRequest.html)
that names the store to query:
`StateQueryRequest.inStore(storeName).withQuery(query)`. Optionally configure
the request (partitions, position bound, isolation level, and so on).
+ 3. **Execute the request.** Call `streams.query(request)`, which runs the
query against every locally available partition of the store (or just the
partitions you requested) and returns a `StateQueryResult`.
+ 4. **Read the results.** For a query that targets a single partition, use
`getOnlyPartitionResult()`. For a query that may span multiple partitions, use
`getPartitionResults()` to get a `Map` from partition number to `QueryResult`.
Each `QueryResult` reports whether the query succeeded on that partition and
holds either the result value or a typed failure reason.
+
+## Building and executing a query
+
+The following example looks up a single key in the `CountsKeyValueStore` state
store from the word-count example used earlier on this page:
+
+```java
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.query.KeyQuery;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.StateQueryRequest;
+import org.apache.kafka.streams.query.StateQueryResult;
+
+import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
+
+KafkaStreams streams = ...;
+
+// 1. Build the query: retrieve the value for a single key.
+KeyQuery<String, Long> query = KeyQuery.withKey("alice");
+
+// 2. Build the request, naming the store to run the query against.
+// A KeyQuery<String, Long> is a Query<Long>, so the request is a
StateQueryRequest<Long>.
+StateQueryRequest<Long> request =
inStore("CountsKeyValueStore").withQuery(query);
+
+// 3. Execute the query.
+StateQueryResult<Long> result = streams.query(request);
+
+// 4. Read the result. A given key lives in exactly one partition, so at most
one partition
+// returns a value. getOnlyPartitionResult() returns that partition's
result, or null if
+// no locally available partition holds the key.
+QueryResult<Long> partitionResult = result.getOnlyPartitionResult();
+if (partitionResult != null && partitionResult.isSuccess()) {
+ Long count = partitionResult.getResult();
+ System.out.println("Count for alice: " + count);
+}
+```
+
+Note that the type parameter of `StateQueryRequest` (and of `StateQueryResult`
and `QueryResult`) is the query's *result* type, not the query type: a
`KeyQuery<String, Long>` implements `Query<Long>`, so the request is a
`StateQueryRequest<Long>`.
+
+## Built-in query types
+
+Kafka Streams ships with a set of query types covering the standard store
types. All of them live in the
[org.apache.kafka.streams.query](/{version}/javadoc/org/apache/kafka/streams/query/package-summary.html)
package.
+
+| Query | Result type (`R`) | Key factory / builder methods |
+| --- | --- | --- |
+| `KeyQuery<K, V>` | `V` | `withKey(key)`; `.skipCache()` |
+| `TimestampedKeyQuery<K, V>` | `ValueAndTimestamp<V>` | `withKey(key)`;
`.skipCache()` |
+| `RangeQuery<K, V>` | `KeyValueIterator<K, V>` | `withRange(lower, upper)`,
`withLowerBound(lower)`, `withUpperBound(upper)`, `withNoBounds()`;
`.withAscendingKeys()` / `.withDescendingKeys()` |
+| `TimestampedRangeQuery<K, V>` | `KeyValueIterator<K, ValueAndTimestamp<V>>`
| `withRange(lower, upper)`, `withLowerBound(lower)`, `withUpperBound(upper)`,
`withNoBounds()`; `.withAscendingKeys()` / `.withDescendingKeys()` |
+| `WindowKeyQuery<K, V>` | `WindowStoreIterator<V>` |
`withKeyAndWindowStartRange(key, timeFrom, timeTo)` |
+| `WindowRangeQuery<K, V>` | `KeyValueIterator<Windowed<K>, V>` |
`withKey(key)`, `withWindowStartRange(timeFrom, timeTo)` |
+| `VersionedKeyQuery<K, V>` | `VersionedRecord<V>` | `withKey(key)`;
`.asOf(instant)` |
+| `MultiVersionedKeyQuery<K, V>` | `VersionedRecordIterator<V>` |
`withKey(key)`; `.fromTime(instant)`, `.toTime(instant)`,
`.withAscendingTimestamps()` / `.withDescendingTimestamps()` |
Review Comment:
[completeness — amends my earlier comment on this line] My earlier comment
asked you to add the four KIP-1356 headers-aware query types to this table.
That still stands, but the recommendation needs qualifying now that I have
looked at **#22882**: that PR documents exactly those four types, in this same
file, in full detail (result semantics, the `*WithHeaders` supplier
requirement, iterator-closing rules, per-store form restrictions).
So please **don't** write that content again here. Better: add the four rows
to this table for discoverability and point into #22882's `## Header-aware
stores and interactive queries` section for the detail — or leave them out of
the table entirely and link to that section from the paragraph below. Which way
round depends on the merge order; see the coordination comment on the section
heading.
##########
docs/streams/developer-guide/interactive-queries.md:
##########
@@ -499,6 +499,198 @@ At this point the full state of the application is
interactively queryable:
+# Interactive Queries v2 (IQv2) {#interactive-queries-v2}
Review Comment:
[coordination] This PR and **#22882** ("KAFKA-20682: Document IQv2
headers-aware queries (KIP-1356)", same author, opened one day earlier) both
add IQv2 documentation to *this same file*, and neither references the other.
They overlap in a way git will not catch:
- **Both introduce IQv2 independently.** #22882's `### Reading headers with
the IQv2 query() API` opens with its own IQv2 primer — build a `Query`, wrap it
in a `StateQueryRequest`, run `KafkaStreams#query(...)`, read
`getOnlyPartitionResult()` / `getPartitionResults()` / `getResult()` /
`getPosition()`. That is the same ground as this PR's "How Interactive Queries
v2 works" and "Handling query results" sections.
- **The reading order would be backwards.** #22882 inserts at ~line 375 (a
`##` under `# Querying local state stores...`); this PR appends at line 502,
*after* `# Querying remote state stores for the entire app`. Land both as-is
and the reader meets headers-specific IQv2 detail roughly 130 lines before IQv2
is introduced at all.
- **The overlap is editorial, not textual.** Both PRs currently report
`MERGEABLE` because their hunks sit at different offsets, so both can merge
cleanly and still leave the page with two IQv2 introductions in the wrong
order. Nothing will warn you.
Both PRs are drafts today, so there is room to sequence this. Suggested
resolution: land #22882 (smaller, self-contained, scoped to one KIP), then
rebase this PR to (a) move this section up to follow `# Querying local state
stores for an app instance` — which is what my `[order]` comment asks for
anyway, and which places it *before* the header-aware section — and (b) reduce
#22882's now-redundant IQv2 primer to a link into this section. The alternative
is to combine the two into a single coherent restructure of the page.
Either way, worth deciding the split before further review rounds on either
PR.
--
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]
