aliehsaeedii commented on code in PR #22896:
URL: https://github.com/apache/kafka/pull/22896#discussion_r3711254891
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
+ * `getPosition()` returns the merged `Position` observed across the
partition results.
+
+Each `QueryResult` reports the outcome for one partition. Use `isSuccess()` /
`isFailure()` before reading: `getResult()` returns the value on success (which
may itself be `null`, for example when a key is not found), while
`getFailureReason()` and `getFailureMessage()` describe a failure. Results are
always **per-partition** — a query may succeed on some partitions and fail on
others (for example, if one partition has migrated off this instance).
+
+When a query returns an iterator (`RangeQuery`, `WindowKeyQuery`,
`MultiVersionedKeyQuery`, and so on), the iterator must be closed after use.
Iterate over every partition's result and use a try-with-resources block:
+
+```java
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.RangeQuery;
+import org.apache.kafka.streams.query.StateQueryRequest;
+import org.apache.kafka.streams.query.StateQueryResult;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.Map;
+
+import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
+
+RangeQuery<String, Long> query = RangeQuery.withRange("a", "n");
+StateQueryRequest<KeyValueIterator<String, Long>> request =
+ inStore("CountsKeyValueStore").withQuery(query);
+StateQueryResult<KeyValueIterator<String, Long>> result =
streams.query(request);
+
+for (Map.Entry<Integer, QueryResult<KeyValueIterator<String, Long>>> entry
+ : result.getPartitionResults().entrySet()) {
+ QueryResult<KeyValueIterator<String, Long>> partitionResult =
entry.getValue();
+ if (partitionResult.isFailure()) {
+ System.out.println("Partition " + entry.getKey() + " failed: "
+ + partitionResult.getFailureReason() + " - " +
partitionResult.getFailureMessage());
+ continue;
+ }
+ try (KeyValueIterator<String, Long> iterator =
partitionResult.getResult()) {
+ while (iterator.hasNext()) {
+ KeyValue<String, Long> record = iterator.next();
+ System.out.println(record.key + ": " + record.value);
+ }
+ }
+}
+```
+
+A failed `QueryResult` carries one of the
[FailureReason](/{version}/javadoc/org/apache/kafka/streams/query/FailureReason.html)
values:
+
+| `FailureReason` | Meaning | Recommended action |
+| --- | --- | --- |
+| `UNKNOWN_QUERY_TYPE` | The store does not know how to execute this query. |
Verify the query is supported by that store type; for custom queries, contact
the store maintainer. |
+| `NOT_ACTIVE` | `requireActive()` was set, but the partition is a standby or
an active task that is not yet in the `RUNNING` state. | Retry later or query a
different replica. |
+| `NOT_UP_TO_BOUND` | The partition has not yet caught up to the requested
`PositionBound`. | Retry later or query a different replica. |
+| `NOT_PRESENT` | The requested partition is not present on this instance (for
example, it migrated during a rebalance). | Query a different replica. |
+| `DOES_NOT_EXIST` | The requested partition does not exist for this store. |
Correct the requested set of partitions. |
+| `STORE_EXCEPTION` | The store threw an exception while executing the query.
| Depending on the exception, retry this instance or a different one. |
+
+## Controlling consistency and availability
+
+A `StateQueryRequest` is immutable; each configuration method returns a new
request. Beyond `inStore(...).withQuery(...)`, the following options let you
trade off consistency, availability, and cost:
+
+ * `withPartitions(Set<Integer>)` / `withAllPartitions()`: run against a
specific set of partitions or against all locally available partitions (the
default). Partitions that are missing return `NOT_PRESENT`; partitions that do
not exist return `DOES_NOT_EXIST`.
+ * `requireActive()`: run only on active (leader) partitions. Non-active
partitions return `NOT_ACTIVE`. Use this when you need the most up-to-date data
and want to avoid reading from standby replicas.
+ * `withPositionBound(PositionBound)`: by default a request is
`PositionBound.unbounded()`. Use `PositionBound.at(position)` to require that
each queried partition has consumed up to a given `Position` before serving the
query; a partition that is behind returns `NOT_UP_TO_BOUND`. Combined with the
`Position` returned by `StateQueryResult#getPosition()`, this lets you
implement read-your-writes / monotonic reads: feed the position from one query
into the bound of the next so repeated queries never appear to move backwards
in time, while still allowing reads to be served from any replica that is
caught up.
+ * `withIsolationLevel(IsolationLevel)`: override the isolation level for
this query. When not set, the effective level falls back to the
`default.interactive.query.isolation.level` configuration.
+ * `enableExecutionInfo()`: ask stores and the runtime to record details
about how the query executed, retrievable via `QueryResult#getExecutionInfo()`.
+
+For example, the following request reads the latest committed value for a key
from the active replica, but only for partitions 0 and 1, and only once the
store has caught up to a known input position:
Review Comment:
[content] Two claims in this sentence don't hold for the example below it.
1. **"only once the store has caught up to a known input position"** — the
example sets both `.requireActive()` and
`.withPositionBound(PositionBound.at(inputPosition))`, but `KafkaStreams#query`
throws the bound away when `requireActive` is set:
```java
final QueryResult<R> r = store.query(
request.getQuery(),
request.isRequireActive()
? PositionBound.unbounded()
: request.getPositionBound(),
```
(`KafkaStreams.java:2219-2224`). So `withPositionBound(...)` has **no
effect** here and `NOT_UP_TO_BOUND` can never be returned.
2. **"reads the latest committed value"** — the request never calls
`withIsolationLevel(...)`, so it inherits
`default.interactive.query.isolation.level`, whose default is
`READ_UNCOMMITTED` (`StreamsConfig.java:553`). The read is not committed-only.
Suggest splitting the example into two (one showing `requireActive()`, one
showing `withPositionBound()` for monotonic reads), and dropping "latest
committed" unless the example adds
`.withIsolationLevel(IsolationLevel.READ_COMMITTED)`.
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
Review Comment:
[content] "filters to the single partition that produced a value" and
"throws `IllegalArgumentException` if more than one partition **produced a
value**" understate when this throws. The filter keeps failures as well as
values:
```java
final List<QueryResult<R>> nonempty =
partitionResults.values().stream()
.filter(r -> r.isFailure() || r.getResult() != null)
.collect(Collectors.toList());
if (nonempty.size() > 1) {
throw new IllegalArgumentException(...)
```
(`StateQueryResult.java:66-81`)
So it throws when more than one partition returns a value **or a failure**.
That matters for the pattern this section recommends: a `KeyQuery` issued with
`requireActive()` — or against an instance hosting several partitions of the
store during a rebalance — yields multiple `NOT_ACTIVE` / `NOT_PRESENT`
failures, and `getOnlyPartitionResult()` then throws instead of returning the
key's result.
Suggest: "returns the single partition result that is either a failure or a
non-`null` value, or `null` if there is none; throws `IllegalArgumentException`
if more than one partition returned a value or a failure." A one-line caution
that failures count towards that limit would save readers a surprise.
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
+ * `getPosition()` returns the merged `Position` observed across the
partition results.
Review Comment:
[completeness] The accessor list is missing `getGlobalResult()`, and its
absence interacts badly with the global-store limitation stated at line 516 and
line 690.
Querying a global store does **not** throw and does **not** populate the
partition results. `KafkaStreams#query` calls `result.setGlobalResult(...)`
with a failed result:
```java
if (globalStateStores.containsKey(storeName)) {
// See KAFKA-13523
result.setGlobalResult(
QueryResult.forFailure(
FailureReason.UNKNOWN_QUERY_TYPE,
"Global stores do not yet support the KafkaStreams#query API.
Use KafkaStreams#store instead."
```
(`KafkaStreams.java:2183-2191`)
So a reader who follows this section verbatim sees an empty
`getPartitionResults()` map and `null` from `getOnlyPartitionResult()`, with
the explanatory failure invisible in `getGlobalResult()`. Suggest a fourth
bullet: "`getGlobalResult()` returns the result for a global-store query and is
`null` for partitioned stores (conversely, `getPartitionResults()` is empty for
global-store queries). This is where the current global-store rejection
surfaces, as a failed result with `UNKNOWN_QUERY_TYPE`."
##########
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`.
Review Comment:
[completeness] This step correctly scopes `query()` to "every locally
available partition", but the section never says how IQv2 relates to
cross-instance / remote querying. A reader arriving from `# Querying remote
state stores for the entire app` will reasonably wonder whether instance
discovery (`StreamsMetadata`, `queryMetadataForKey`) applies to IQv2 or whether
`query()` transparently fans out across instances.
A one-line note — that `query()` only reads this instance's local state, and
that discovering which instance hosts a key works the same way as for IQv1 —
would close the gap and tie the two halves of the page together.
(Carried forward from an earlier pending review, now line-anchored.)
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
+ * `getPosition()` returns the merged `Position` observed across the
partition results.
+
+Each `QueryResult` reports the outcome for one partition. Use `isSuccess()` /
`isFailure()` before reading: `getResult()` returns the value on success (which
may itself be `null`, for example when a key is not found), while
`getFailureReason()` and `getFailureMessage()` describe a failure. Results are
always **per-partition** — a query may succeed on some partitions and fail on
others (for example, if one partition has migrated off this instance).
+
+When a query returns an iterator (`RangeQuery`, `WindowKeyQuery`,
`MultiVersionedKeyQuery`, and so on), the iterator must be closed after use.
Iterate over every partition's result and use a try-with-resources block:
+
+```java
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.RangeQuery;
+import org.apache.kafka.streams.query.StateQueryRequest;
+import org.apache.kafka.streams.query.StateQueryResult;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.Map;
+
+import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
+
+RangeQuery<String, Long> query = RangeQuery.withRange("a", "n");
+StateQueryRequest<KeyValueIterator<String, Long>> request =
+ inStore("CountsKeyValueStore").withQuery(query);
+StateQueryResult<KeyValueIterator<String, Long>> result =
streams.query(request);
+
+for (Map.Entry<Integer, QueryResult<KeyValueIterator<String, Long>>> entry
+ : result.getPartitionResults().entrySet()) {
+ QueryResult<KeyValueIterator<String, Long>> partitionResult =
entry.getValue();
+ if (partitionResult.isFailure()) {
+ System.out.println("Partition " + entry.getKey() + " failed: "
+ + partitionResult.getFailureReason() + " - " +
partitionResult.getFailureMessage());
+ continue;
+ }
+ try (KeyValueIterator<String, Long> iterator =
partitionResult.getResult()) {
+ while (iterator.hasNext()) {
+ KeyValue<String, Long> record = iterator.next();
+ System.out.println(record.key + ": " + record.value);
+ }
+ }
+}
+```
+
+A failed `QueryResult` carries one of the
[FailureReason](/{version}/javadoc/org/apache/kafka/streams/query/FailureReason.html)
values:
+
+| `FailureReason` | Meaning | Recommended action |
+| --- | --- | --- |
+| `UNKNOWN_QUERY_TYPE` | The store does not know how to execute this query. |
Verify the query is supported by that store type; for custom queries, contact
the store maintainer. |
+| `NOT_ACTIVE` | `requireActive()` was set, but the partition is a standby or
an active task that is not yet in the `RUNNING` state. | Retry later or query a
different replica. |
+| `NOT_UP_TO_BOUND` | The partition has not yet caught up to the requested
`PositionBound`. | Retry later or query a different replica. |
+| `NOT_PRESENT` | The requested partition is not present on this instance (for
example, it migrated during a rebalance). | Query a different replica. |
+| `DOES_NOT_EXIST` | The requested partition does not exist for this store. |
Correct the requested set of partitions. |
+| `STORE_EXCEPTION` | The store threw an exception while executing the query.
| Depending on the exception, retry this instance or a different one. |
+
+## Controlling consistency and availability
+
+A `StateQueryRequest` is immutable; each configuration method returns a new
request. Beyond `inStore(...).withQuery(...)`, the following options let you
trade off consistency, availability, and cost:
+
+ * `withPartitions(Set<Integer>)` / `withAllPartitions()`: run against a
specific set of partitions or against all locally available partitions (the
default). Partitions that are missing return `NOT_PRESENT`; partitions that do
not exist return `DOES_NOT_EXIST`.
+ * `requireActive()`: run only on active (leader) partitions. Non-active
partitions return `NOT_ACTIVE`. Use this when you need the most up-to-date data
and want to avoid reading from standby replicas.
+ * `withPositionBound(PositionBound)`: by default a request is
`PositionBound.unbounded()`. Use `PositionBound.at(position)` to require that
each queried partition has consumed up to a given `Position` before serving the
query; a partition that is behind returns `NOT_UP_TO_BOUND`. Combined with the
`Position` returned by `StateQueryResult#getPosition()`, this lets you
implement read-your-writes / monotonic reads: feed the position from one query
into the bound of the next so repeated queries never appear to move backwards
in time, while still allowing reads to be served from any replica that is
caught up.
+ * `withIsolationLevel(IsolationLevel)`: override the isolation level for
this query. When not set, the effective level falls back to the
`default.interactive.query.isolation.level` configuration.
Review Comment:
[completeness] The config name and fallback behaviour are both correct
(`StreamsConfig.java:547`; `StateQueryRequest.java:151-153`, and the runtime
resolves it via
`request.isolationLevel().orElseGet(applicationConfigs::defaultInteractiveQueryIsolationLevel)`
at `KafkaStreams.java:2225-2228`). What's missing is the precondition that
makes the option meaningful — the config's own documentation says:
> Only meaningful when `enable.transactional.statestores` is `true`:
`READ_UNCOMMITTED` reads include writes staged in the transaction buffer since
the last commit; `READ_COMMITTED` reads skip the transaction buffer and return
only committed data.
(`StreamsConfig.java:548-552`)
Also worth stating the default, `READ_UNCOMMITTED`
(`StreamsConfig.java:553`), since a reader may assume IQ defaults to committed
reads. A short parenthetical covering both would make this bullet
self-contained.
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
+ * `getPosition()` returns the merged `Position` observed across the
partition results.
+
+Each `QueryResult` reports the outcome for one partition. Use `isSuccess()` /
`isFailure()` before reading: `getResult()` returns the value on success (which
may itself be `null`, for example when a key is not found), while
`getFailureReason()` and `getFailureMessage()` describe a failure. Results are
always **per-partition** — a query may succeed on some partitions and fail on
others (for example, if one partition has migrated off this instance).
+
+When a query returns an iterator (`RangeQuery`, `WindowKeyQuery`,
`MultiVersionedKeyQuery`, and so on), the iterator must be closed after use.
Iterate over every partition's result and use a try-with-resources block:
+
+```java
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.RangeQuery;
+import org.apache.kafka.streams.query.StateQueryRequest;
+import org.apache.kafka.streams.query.StateQueryResult;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.Map;
+
+import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
+
+RangeQuery<String, Long> query = RangeQuery.withRange("a", "n");
+StateQueryRequest<KeyValueIterator<String, Long>> request =
+ inStore("CountsKeyValueStore").withQuery(query);
+StateQueryResult<KeyValueIterator<String, Long>> result =
streams.query(request);
+
+for (Map.Entry<Integer, QueryResult<KeyValueIterator<String, Long>>> entry
+ : result.getPartitionResults().entrySet()) {
+ QueryResult<KeyValueIterator<String, Long>> partitionResult =
entry.getValue();
+ if (partitionResult.isFailure()) {
+ System.out.println("Partition " + entry.getKey() + " failed: "
+ + partitionResult.getFailureReason() + " - " +
partitionResult.getFailureMessage());
+ continue;
+ }
+ try (KeyValueIterator<String, Long> iterator =
partitionResult.getResult()) {
+ while (iterator.hasNext()) {
+ KeyValue<String, Long> record = iterator.next();
+ System.out.println(record.key + ": " + record.value);
+ }
+ }
+}
+```
+
+A failed `QueryResult` carries one of the
[FailureReason](/{version}/javadoc/org/apache/kafka/streams/query/FailureReason.html)
values:
+
+| `FailureReason` | Meaning | Recommended action |
+| --- | --- | --- |
+| `UNKNOWN_QUERY_TYPE` | The store does not know how to execute this query. |
Verify the query is supported by that store type; for custom queries, contact
the store maintainer. |
+| `NOT_ACTIVE` | `requireActive()` was set, but the partition is a standby or
an active task that is not yet in the `RUNNING` state. | Retry later or query a
different replica. |
+| `NOT_UP_TO_BOUND` | The partition has not yet caught up to the requested
`PositionBound`. | Retry later or query a different replica. |
+| `NOT_PRESENT` | The requested partition is not present on this instance (for
example, it migrated during a rebalance). | Query a different replica. |
+| `DOES_NOT_EXIST` | The requested partition does not exist for this store. |
Correct the requested set of partitions. |
+| `STORE_EXCEPTION` | The store threw an exception while executing the query.
| Depending on the exception, retry this instance or a different one. |
+
+## Controlling consistency and availability
+
+A `StateQueryRequest` is immutable; each configuration method returns a new
request. Beyond `inStore(...).withQuery(...)`, the following options let you
trade off consistency, availability, and cost:
+
+ * `withPartitions(Set<Integer>)` / `withAllPartitions()`: run against a
specific set of partitions or against all locally available partitions (the
default). Partitions that are missing return `NOT_PRESENT`; partitions that do
not exist return `DOES_NOT_EXIST`.
+ * `requireActive()`: run only on active (leader) partitions. Non-active
partitions return `NOT_ACTIVE`. Use this when you need the most up-to-date data
and want to avoid reading from standby replicas.
+ * `withPositionBound(PositionBound)`: by default a request is
`PositionBound.unbounded()`. Use `PositionBound.at(position)` to require that
each queried partition has consumed up to a given `Position` before serving the
query; a partition that is behind returns `NOT_UP_TO_BOUND`. Combined with the
`Position` returned by `StateQueryResult#getPosition()`, this lets you
implement read-your-writes / monotonic reads: feed the position from one query
into the bound of the next so repeated queries never appear to move backwards
in time, while still allowing reads to be served from any replica that is
caught up.
Review Comment:
[content] This bullet needs the `requireActive()` interaction spelled out,
because it silently defeats the read-your-writes recipe described here.
`KafkaStreams#query` substitutes `PositionBound.unbounded()` for the request's
bound whenever `requireActive` is set (`KafkaStreams.java:2221-2224`), so a
request that combines the two ignores the bound entirely.
Suggested addition: "`withPositionBound(...)` is ignored when
`requireActive()` is also set — an active, running task is served without a
bound check." (This is what makes the example below at line 659 incorrect.)
##########
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.
Review Comment:
[completeness] "typed failure reasons instead of exceptions" is right about
*per-partition* outcomes, but `query()` itself still throws, and the section
never says so. Per the `KafkaStreams#query` javadoc and its preconditions
(`KafkaStreams.java:2153-2180`):
- `UnknownStateStoreException` — the store name is not registered in the
topology
- `StreamsNotStartedException` — `start()` hasn't been called yet; retry
after starting
- `StreamsStoppedException` — the instance is in `PENDING_SHUTDOWN` /
`NOT_RUNNING` / `PENDING_ERROR` / `ERROR`; discover a new instance
Worth one line here (or in "Handling query results") so readers understand
the split: request-level problems raise exceptions, per-partition problems come
back as `FailureReason`s. Otherwise "instead of exceptions" reads as though
`query()` never throws.
##########
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)` |
Review Comment:
[content] Listing `withKey(key)` and `withWindowStartRange(timeFrom,
timeTo)` on one row implies both work against the same store. They don't — each
is accepted by a *different* store type, and the other fails:
- **Window stores** support only `withWindowStartRange`.
`MeteredWindowStore#runRangeQuery` returns `UNKNOWN_QUERY_TYPE` otherwise, with
the message `"WindowStores only supports
WindowRangeQuery.withWindowStartRange."`
- **Session stores** support only `withKey`. `MeteredSessionStore` registers
exactly one handler, for `WindowRangeQuery` (`MeteredSessionStore.java:86-92`),
and its `runRangeQuery` fails anything else with `"SessionStores only support
WindowRangeQuery.withKey."`
Suggest splitting into two rows (or adding a store column), e.g.
`WindowRangeQuery` / window store / `withWindowStartRange(timeFrom, timeTo)`
and `WindowRangeQuery` / session store / `withKey(key)`.
This is also the only place session stores could be mentioned — worth
stating explicitly that session stores *are* IQv2-queryable via
`WindowRangeQuery.withKey`, since a reader scanning this table would otherwise
conclude they aren't supported at all.
##########
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:
[order] This new top-level `#` section is appended after `# Querying remote
state stores for the entire app` (line 377), but the section itself correctly
frames IQv2 as an alternative to the *local* querying API introduced in `#
Querying local state stores for an app instance` (line 126). Reading
top-to-bottom, a user meets remote querying and the RPC layer before learning
that IQv2 exists as a local alternative.
Consider moving this section to directly follow the local-querying section,
or adding a one-line forward pointer to it from there. The section is
self-contained, so this is a flow preference rather than a correctness issue.
(Carried forward from an earlier pending review, now line-anchored.)
##########
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] The table is introduced as the set of query types Kafka
Streams ships with, but four public query types in
`org.apache.kafka.streams.query` are missing — the KIP-1356 headers-aware
queries. All four are `@Evolving @InterfaceAudience.Public`, and all landed
before this PR was opened (2026-07-21):
| Query | Result type (`R`) | Commit |
| --- | --- | --- |
| `TimestampedKeyWithHeadersQuery<K, V>` | `ReadOnlyRecord<K, V>` |
`eb7096dcf6` (#22666) |
| `TimestampedRangeWithHeadersQuery<K, V>` | `ReadOnlyRecordIterator<K, V>`
| `a49d127f5b` (#22770) |
| `TimestampedWindowKeyWithHeadersQuery<K, V>` |
`ReadOnlyRecordIterator<Windowed<K>, V>` | `cacab1cc81` (#22799) |
| `TimestampedWindowRangeWithHeadersQuery<K, V>` |
`ReadOnlyRecordIterator<Windowed<K>, V>` | `8001408711` (#22853) |
Their factory methods mirror the non-headers variants (`withKey`,
`withRange` / `withLowerBound` / `withUpperBound` / `withNoBounds`,
`withKeyAndWindowStartRange`, `withWindowStartRange`). Since they require a
headers-aware store supplier, a short note to that effect would help. Worth
adding rows so the table doesn't go stale on arrival.
##########
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()` |
+
+For range and scan queries (`RangeQuery`, `TimestampedRangeQuery`), passing no
bounds performs a full scan, and result ordering is based on the serialized
`byte[]` of the keys, not on the logical key order.
+
+Versioned key-value stores are queryable **only** through IQv2 — use
`VersionedKeyQuery` for a single version (latest, or as of a timestamp) and
`MultiVersionedKeyQuery` for a range of versions. The original
`KafkaStreams#store(...)` API has no queryable store type for versioned stores.
+
+Because IQv2 is extensible, a custom state store may implement additional
query types of its own. When a store does not know how to handle a query, it
does not throw; instead it returns a failed `QueryResult` with
`FailureReason.UNKNOWN_QUERY_TYPE`.
+
+## Handling query results
+
+A `StateQueryResult` aggregates one `QueryResult` per partition that ran the
query. By default a request runs against all locally available partitions of
the store (unless you narrow it with `withPartitions(...)`), so
`getPartitionResults()` may contain an entry for every one of those partitions.
+
+Which accessor to use is determined by the *query type* you chose, not by
inspecting the result at runtime:
+
+ * **Point lookups** (`KeyQuery`, `TimestampedKeyQuery`, `VersionedKeyQuery`)
can only match in the single partition that owns the key, so at most one
partition returns a value (the other queried partitions return a successful
result with `null`). Use `getOnlyPartitionResult()`.
+ * **Range, scan, and window queries** (`RangeQuery`, `WindowRangeQuery`,
`MultiVersionedKeyQuery`, and so on) can match records in every queried
partition, so results are spread across partitions. Use `getPartitionResults()`
and iterate.
+
+The accessors are:
+
+ * `getPartitionResults()` returns a `Map<Integer, QueryResult<R>>`, keyed by
partition number — one entry per partition that ran the query. This is the
general form and works for any query.
+ * `getOnlyPartitionResult()` is a convenience that filters to the single
partition that produced a value and returns it (or `null` if none did). It
throws `IllegalArgumentException` if more than one partition produced a value,
so only use it when you know the query matches at most one partition.
+ * `getPosition()` returns the merged `Position` observed across the
partition results.
+
+Each `QueryResult` reports the outcome for one partition. Use `isSuccess()` /
`isFailure()` before reading: `getResult()` returns the value on success (which
may itself be `null`, for example when a key is not found), while
`getFailureReason()` and `getFailureMessage()` describe a failure. Results are
always **per-partition** — a query may succeed on some partitions and fail on
others (for example, if one partition has migrated off this instance).
+
+When a query returns an iterator (`RangeQuery`, `WindowKeyQuery`,
`MultiVersionedKeyQuery`, and so on), the iterator must be closed after use.
Iterate over every partition's result and use a try-with-resources block:
+
+```java
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.RangeQuery;
+import org.apache.kafka.streams.query.StateQueryRequest;
+import org.apache.kafka.streams.query.StateQueryResult;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.Map;
+
+import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
+
+RangeQuery<String, Long> query = RangeQuery.withRange("a", "n");
+StateQueryRequest<KeyValueIterator<String, Long>> request =
+ inStore("CountsKeyValueStore").withQuery(query);
+StateQueryResult<KeyValueIterator<String, Long>> result =
streams.query(request);
+
+for (Map.Entry<Integer, QueryResult<KeyValueIterator<String, Long>>> entry
+ : result.getPartitionResults().entrySet()) {
+ QueryResult<KeyValueIterator<String, Long>> partitionResult =
entry.getValue();
+ if (partitionResult.isFailure()) {
+ System.out.println("Partition " + entry.getKey() + " failed: "
+ + partitionResult.getFailureReason() + " - " +
partitionResult.getFailureMessage());
+ continue;
+ }
+ try (KeyValueIterator<String, Long> iterator =
partitionResult.getResult()) {
+ while (iterator.hasNext()) {
+ KeyValue<String, Long> record = iterator.next();
+ System.out.println(record.key + ": " + record.value);
+ }
+ }
+}
+```
+
+A failed `QueryResult` carries one of the
[FailureReason](/{version}/javadoc/org/apache/kafka/streams/query/FailureReason.html)
values:
+
+| `FailureReason` | Meaning | Recommended action |
+| --- | --- | --- |
+| `UNKNOWN_QUERY_TYPE` | The store does not know how to execute this query. |
Verify the query is supported by that store type; for custom queries, contact
the store maintainer. |
+| `NOT_ACTIVE` | `requireActive()` was set, but the partition is a standby or
an active task that is not yet in the `RUNNING` state. | Retry later or query a
different replica. |
+| `NOT_UP_TO_BOUND` | The partition has not yet caught up to the requested
`PositionBound`. | Retry later or query a different replica. |
+| `NOT_PRESENT` | The requested partition is not present on this instance (for
example, it migrated during a rebalance). | Query a different replica. |
+| `DOES_NOT_EXIST` | The requested partition does not exist for this store. |
Correct the requested set of partitions. |
+| `STORE_EXCEPTION` | The store threw an exception while executing the query.
| Depending on the exception, retry this instance or a different one. |
+
+## Controlling consistency and availability
+
+A `StateQueryRequest` is immutable; each configuration method returns a new
request. Beyond `inStore(...).withQuery(...)`, the following options let you
trade off consistency, availability, and cost:
+
+ * `withPartitions(Set<Integer>)` / `withAllPartitions()`: run against a
specific set of partitions or against all locally available partitions (the
default). Partitions that are missing return `NOT_PRESENT`; partitions that do
not exist return `DOES_NOT_EXIST`.
Review Comment:
[content] "partitions that do not exist return `DOES_NOT_EXIST`" is not what
the runtime does. `FailureReason.DOES_NOT_EXIST` is described in the
`StateQueryRequest#withPartitions` javadoc (`StateQueryRequest.java:103-105`),
but no production code ever constructs it — `grep -rn DOES_NOT_EXIST
streams/src/main/java` only hits the enum declaration and that javadoc, and the
sole other occurrence in the repo is a unit test
(`StateQueryResultTest.java:36`).
What actually happens: after the task scan, `KafkaStreams#query` fills in
*every* requested-but-unresolved partition with `NOT_PRESENT`
(`KafkaStreams.java:2245-2253`), whether the partition merely migrated away or
does not exist for the store at all.
Same issue for the `DOES_NOT_EXIST` row of the `FailureReason` table at line
646. Suggest either dropping the claim that invalid partitions produce
`DOES_NOT_EXIST` (say `NOT_PRESENT` is returned for both cases) or noting that
`DOES_NOT_EXIST` is defined for store implementations but is not currently
emitted by the Streams runtime.
--
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]
