This is an automated email from the ASF dual-hosted git repository.
frankvicky pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a49d127f5b9 KAFKA-20704: IQv2 TimestampedRangeWithHeadersQuery for
headers-aware key-value stores (KIP-1356) (#22770)
a49d127f5b9 is described below
commit a49d127f5b901e9b93391f75d8efedc12857d6cd
Author: Jess668 <[email protected]>
AuthorDate: Tue Jul 14 04:31:51 2026 -0400
KAFKA-20704: IQv2 TimestampedRangeWithHeadersQuery for headers-aware
key-value stores (KIP-1356) (#22770)
Implements **`TimestampedRangeWithHeadersQuery`**, the second IQv2 query
type of KIP-1356, whose result is an iterator of records that each carry
key, value, timestamp, **and headers**. The result type is the
**`ReadOnlyRecordIterator<K, V>`** — a `Closeable` iterator that yields
`ReadOnlyRecord<K, V>` elements directly.
## Dependencies
This work is stacked on, and should be reviewed/merged after:
1. `ReadOnlyRecord`
[PR#22677](https://github.com/apache/kafka/pull/22677) — already in
`trunk`.
2. `ReadOnlyRecordIterator`
[PR#22729](https://github.com/apache/kafka/pull/22729) — already in
`trunk`.
3. `TimestampedKeyWithHeadersQuery`
[PR#22666](https://github.com/apache/kafka/pull/22666) — already in
`trunk`.
All three have landed; this branch has been rebased onto `trunk` and
contains only the range commits.
## Summary
1. **`TimestampedRangeWithHeadersQuery<K, V> implements
Query<ReadOnlyRecordIterator<K, V>>`** (`@Evolving` /
`@InterfaceAudience.Public`). Mirrors `TimestampedRangeQuery`: static
factories `withRange`, `withUpperBound`, `withLowerBound`,
`withNoBounds`; order builders `withDescendingKeys` /
`withAscendingKeys`; accessors `lowerBound()`, `upperBound()`,
`resultOrder()`. Same bound/order semantics as `TimestampedRangeQuery`.
2. `MeteredTimestampedKeyValueStoreWithHeaders` registers a
`TimestampedRangeWithHeadersQuery` handler and returns a
`ReadOnlyRecordIterator` whose elements are built as immutable
`ReadOnlyRecord`s (headers frozen via `setReadOnly()`). The bounds+order
→ raw `RangeQuery` construction shared by the three range handlers is
factored into a `rawRangeQuery(...)` helper.
3. The shared open/close/gauge scaffolding across the three metered
iterator classes in `MeteredTimestampedKeyValueStoreWithHeaders` is
factored into an `AbstractMeteredIterator` base. The
`MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator` now
documents that a caller must still `close()` it in a
`finally`/try-with-resources even when `next()` throws for a
negative-timestamp entry, or the gauge and the underlying iterator both
leak.
4. **`num-open-iterators` metric fix**: the IQ range iterators
(`QueryIterator`, used by `RangeQuery`/`TimestampedRangeQuery`, and the
new `ReadOnlyRecordIterator` for this query) registered in
`openIterators` but never updated `numOpenIterators`, so they were
invisible to the `num-open-iterators` gauge. Both now increment on open
/ decrement on close, matching the base `MeteredKeyValueStoreIterator`.
5. **Native store enablement**: `RocksDBTimestampedStoreWithHeaders`
drops its scope-to-`KeyQuery`-only `query()` override, falling back to
the inherited `RocksDBStore` handling, which serves `RangeQuery`.
Previously the native header store returned `UNKNOWN_QUERY_TYPE` for
range queries.
## Testing
- **Unit** — `MeteredTimestampedKeyValueStoreWithHeadersTest`: bounds
(as serialized keys) and ascending/descending order are propagated to
the underlying raw `RangeQuery`.
- **Builder / end-to-end** —
`TimestampedKeyValueStoreBuilderWithHeadersTest`: `RangeQuery` on the
native store; `TimestampedRangeWithHeadersQuery` returns exact headers
on a header-persisting store, empty headers on the adapter store, throws
on a plain/legacy store, throws on a negative stored timestamp;
execution-info collect/no-collect; identical range results across
native- and adapter-built stores.
- **Native store** — `RocksDBTimestampedStoreWithHeadersTest`:
`RangeQuery` is now handled (was `UNKNOWN_QUERY_TYPE`).
- **Integration** — `IQv2HeadersStoreIntegrationTest`:
`TimestampedRangeWithHeadersQuery` over a running topology returns each
record's correct key/value/timestamp/headers across order and bound
variants; `UNKNOWN_QUERY_TYPE` against a non-headers store; failure
against a plain supplier.
## Follow-ups
- The sibling `MeteredTimestampedKeyValueStore` (non-headers store) has
the same `num-open-iterators` gauge gap fixed here; filing a separate
JIRA [KAFKA-20796](https://issues.apache.org/jira/browse/KAFKA-20796) to
track it rather than fixing it in this PR.
Reviewers: Alieh Saeedi <[email protected]>, TengYao Chi
<[email protected]>
---
.../IQv2HeadersStoreIntegrationTest.java | 346 ++++++++++++++++++---
.../query/TimestampedRangeWithHeadersQuery.java | 156 ++++++++++
...MeteredTimestampedKeyValueStoreWithHeaders.java | 247 ++++++++++-----
.../RocksDBTimestampedStoreWithHeaders.java | 34 --
...redTimestampedKeyValueStoreWithHeadersTest.java | 177 ++++++++++-
.../RocksDBTimestampedStoreWithHeadersTest.java | 27 +-
...stampedKeyValueStoreBuilderWithHeadersTest.java | 251 +++++++++++++--
7 files changed, 1048 insertions(+), 190 deletions(-)
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java
index f8eb62da2e1..9b1442883ca 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java
@@ -30,21 +30,26 @@ import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import org.apache.kafka.streams.processor.api.ReadOnlyRecord;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.query.FailureReason;
import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.query.PositionBound;
+import org.apache.kafka.streams.query.Query;
import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.query.StateQueryRequest;
import org.apache.kafka.streams.query.StateQueryResult;
import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
+import org.apache.kafka.streams.query.TimestampedRangeWithHeadersQuery;
+import org.apache.kafka.streams.state.ReadOnlyRecordIterator;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.TimestampedKeyValueStore;
@@ -63,24 +68,28 @@ import org.junit.jupiter.api.TestInfo;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* IQv2 integration tests for KIP-1271/KIP-1356 headers-aware state stores.
*
* <p>It builds a KIP-1271 {@code WithHeaders} store, writes records (with
headers) into it through a processor,
- * and queries it through IQv2. Currently covers {@link
TimestampedKeyWithHeadersQuery}; the remaining KIP-1356 headers
- * query types (range/window/session) are expected to extend this class as
they land.
+ * and queries it through IQv2. Covers {@link TimestampedKeyWithHeadersQuery}
and
+ * {@link TimestampedRangeWithHeadersQuery}; the remaining KIP-1356 headers
query types (window/session) are
+ * expected to extend this class as they land.
*/
@Tag("integration")
public class IQv2HeadersStoreIntegrationTest {
@@ -133,7 +142,7 @@ public class IQv2HeadersStoreIntegrationTest {
@Test
public void shouldHandleTimestampedKeyWithHeadersQuery() throws Exception {
- startStreams();
+ startStreamsWithKeyValueHeadersStore();
// key 1 has headers, key 2 has empty headers, key 3 is tombstoned
(null value)
produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
@@ -144,24 +153,24 @@ public class IQv2HeadersStoreIntegrationTest {
KeyValue.pair(3, "c0"), KeyValue.pair(3, null));
// key 1: key + value + timestamp + headers round-trip
- final ReadOnlyRecord<Integer, String> result1 = query(1);
+ final ReadOnlyRecord<Integer, String> result1 = keyQuery(1);
assertEquals(Integer.valueOf(1), result1.key());
assertEquals("a0", result1.value());
assertEquals(baseTimestamp, result1.timestamp());
assertEquals(HEADERS, result1.headers());
// key 2: written with no headers -> empty (never null) headers
- final ReadOnlyRecord<Integer, String> result2 = query(2);
+ final ReadOnlyRecord<Integer, String> result2 = keyQuery(2);
assertEquals(Integer.valueOf(2), result2.key());
assertEquals("b0", result2.value());
assertEquals(baseTimestamp + 1, result2.timestamp());
assertEquals(new RecordHeaders(), result2.headers());
// key 3: tombstoned -> null result, never a partially-populated
wrapper
- assertNull(query(3));
+ assertNull(keyQuery(3));
// never-written key -> null result
- assertNull(query(999));
+ assertNull(keyQuery(999));
}
@Test
@@ -171,16 +180,16 @@ public class IQv2HeadersStoreIntegrationTest {
// successful query therefore proves the result was served from the
cache via
// CachingKeyValueStoreWithHeaders -> the metered store's cache-hit
path, end-to-end.
commitIntervalMs = Duration.ofMinutes(10).toMillis();
- startStreams(true);
+ startStreamsWithKeyValueHeadersStore(true);
produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
// Read-your-writes: the not-yet-flushed record is visible (served
from the cache), with headers.
- final ReadOnlyRecord<Integer, String> result = query(1);
+ final ReadOnlyRecord<Integer, String> result = keyQuery(1);
// skipCache bypasses the cache and reads the persistent store
directly. A null result positively
// proves nothing has been flushed, so the read above was genuinely
cache-served (not an accidental
// store read) -- and it covers skipCache end-to-end.
- assertNull(querySkipCache(1));
+ assertNull(keyQuerySkipCache(1));
assertEquals(Integer.valueOf(1), result.key());
assertEquals("a0", result.value());
assertEquals(baseTimestamp, result.timestamp());
@@ -188,60 +197,277 @@ public class IQv2HeadersStoreIntegrationTest {
}
@Test
- public void shouldFailWithUnknownQueryTypeAgainstNonHeadersStore() throws
Exception {
- // store built WITHOUT a WithHeaders supplier -> the query type is
unsupported
- final StreamsBuilder builder = new StreamsBuilder();
- builder
- .addStateStore(
- Stores.timestampedKeyValueStoreBuilder(
- Stores.persistentTimestampedKeyValueStore(STORE_NAME),
- Serdes.Integer(),
- Serdes.String()))
- .stream(inputStream, Consumed.with(Serdes.Integer(),
Serdes.String()))
- .process(() -> new PlainStoreWriterProcessor(), STORE_NAME)
- .to(outputStream, Produced.with(Serdes.Integer(),
Serdes.String()));
+ public void
shouldFailWithUnknownQueryTypeForKeyQueryAgainstNonHeadersStore() throws
Exception {
+
assertUnknownQueryTypeAgainstNonHeadersStore(TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(1));
+ }
- kafkaStreams = new KafkaStreams(builder.build(), props());
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+ @Test
+ public void shouldHandleTimestampedRangeWithHeadersQuery() throws
Exception {
+ // Caching disabled: a range query reads the underlying store directly
(it never consults the
+ // cache), so the writes must be store-served.
+ startStreamsWithKeyValueHeadersStore();
+
+ // keys 1,2 (headers), key 3 (empty headers), key 4 written then
tombstone
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
+ KeyValue.pair(1, "one"), KeyValue.pair(2, "two"));
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp + 1, new
RecordHeaders(),
+ KeyValue.pair(3, "three"));
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp + 2, HEADERS,
+ KeyValue.pair(4, "four"), KeyValue.pair(4, null));
+
+ // Full scan, ascending: keys 1, 2, 3 (key 4 tombstone and omitted).
+ final List<ReadOnlyRecord<Integer, String>> ascending =
+ rangeQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds().withAscendingKeys());
+ assertEquals(List.of(1, 2, 3), keys(ascending));
+ assertEquals(List.of("one", "two", "three"), values(ascending));
+ // key/value/timestamp/headers carried on each element
+ assertEquals(HEADERS, ascending.get(0).headers()); // key 1
+ assertEquals(baseTimestamp, ascending.get(0).timestamp());
+ assertEquals(HEADERS, ascending.get(1).headers()); // key 2
+ assertEquals(baseTimestamp, ascending.get(1).timestamp());
+ // key 3 written without headers -> empty (never null) headers
+ assertEquals(new RecordHeaders(), ascending.get(2).headers());
+ assertEquals(baseTimestamp + 1, ascending.get(2).timestamp());
+ // returned headers are a read-only snapshot: no mutation (add or
remove) is allowed
+ assertThrows(IllegalStateException.class, () ->
ascending.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
ascending.get(0).headers().remove("source"));
+
+ // Full scan, descending: keys reversed, payload still carried.
+ final List<ReadOnlyRecord<Integer, String>> descending =
+ rangeQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds().withDescendingKeys());
+ assertEquals(List.of(3, 2, 1), keys(descending));
+ assertEquals(List.of("three", "two", "one"), values(descending));
+ // key/value/timestamp/headers carried on each element
+ // key 3 written without headers -> empty (never null) headers
+ assertEquals(new RecordHeaders(), descending.get(0).headers()); //
key 3
+ assertEquals(baseTimestamp + 1, descending.get(0).timestamp());
+ assertEquals(HEADERS, descending.get(1).headers()); // key
2
+ assertEquals(baseTimestamp, descending.get(1).timestamp());
+ assertEquals(HEADERS, descending.get(2).headers()); // key
1
+ assertEquals(baseTimestamp, descending.get(2).timestamp());
+ // returned headers are a read-only snapshot: no mutation (add or
remove) is allowed
+ assertThrows(IllegalStateException.class, () ->
descending.get(2).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
descending.get(2).headers().remove("source"));
+
+ // Bounded ranges (inclusive on both ends); same per-element checks as
the full scans.
+ // withRange(2, 3) -> keys 2, 3.
+ final List<ReadOnlyRecord<Integer, String>> range =
rangeQuery(TimestampedRangeWithHeadersQuery.withRange(2, 3));
+ assertEquals(List.of(2, 3), keys(range));
+ assertEquals(List.of("two", "three"), values(range));
+ assertEquals(HEADERS, range.get(0).headers()); // key
2
+ assertEquals(baseTimestamp, range.get(0).timestamp());
+ // key 3 written without headers -> empty (never null) headers
+ assertEquals(new RecordHeaders(), range.get(1).headers()); // key
3
+ assertEquals(baseTimestamp + 1, range.get(1).timestamp());
+ // returned headers are a read-only snapshot: no mutation (add or
remove) is allowed
+ assertThrows(IllegalStateException.class, () ->
range.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
range.get(0).headers().remove("source"));
+
+ // withLowerBound(2) -> keys 2, 3.
+ final List<ReadOnlyRecord<Integer, String>> lowerBounded =
+ rangeQuery(TimestampedRangeWithHeadersQuery.withLowerBound(2));
+ assertEquals(List.of(2, 3), keys(lowerBounded));
+ assertEquals(List.of("two", "three"), values(lowerBounded));
+ assertEquals(HEADERS, lowerBounded.get(0).headers()); // key
2
+ assertEquals(baseTimestamp, lowerBounded.get(0).timestamp());
+ assertEquals(new RecordHeaders(), lowerBounded.get(1).headers()); //
key 3
+ assertEquals(baseTimestamp + 1, lowerBounded.get(1).timestamp());
+ assertThrows(IllegalStateException.class, () ->
lowerBounded.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
lowerBounded.get(0).headers().remove("source"));
+
+ // withUpperBound(2) -> keys 1, 2.
+ final List<ReadOnlyRecord<Integer, String>> upperBounded =
+ rangeQuery(TimestampedRangeWithHeadersQuery.withUpperBound(2));
+ assertEquals(List.of(1, 2), keys(upperBounded));
+ assertEquals(List.of("one", "two"), values(upperBounded));
+ assertEquals(HEADERS, upperBounded.get(0).headers()); // key
1
+ assertEquals(baseTimestamp, upperBounded.get(0).timestamp());
+ assertEquals(HEADERS, upperBounded.get(1).headers()); // key
2
+ assertEquals(baseTimestamp, upperBounded.get(1).timestamp());
+ assertThrows(IllegalStateException.class, () ->
upperBounded.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
upperBounded.get(0).headers().remove("source"));
+
+ // withRange(2, 2), equal bounds -> key 2 only (bounds are inclusive
on both ends).
+ final List<ReadOnlyRecord<Integer, String>> equalBounds =
rangeQuery(TimestampedRangeWithHeadersQuery.withRange(2, 2));
+ assertEquals(List.of(2), keys(equalBounds));
+ assertEquals(List.of("two"), values(equalBounds));
+ assertEquals(HEADERS, equalBounds.get(0).headers());
+ assertEquals(baseTimestamp, equalBounds.get(0).timestamp());
+ assertThrows(IllegalStateException.class, () ->
equalBounds.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
equalBounds.get(0).headers().remove("source"));
+
+ // withRange(2, 3), descending -> keys 3, 2.
+ final List<ReadOnlyRecord<Integer, String>> rangeDescending =
+ rangeQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withRange(2, 3).withDescendingKeys());
+ assertEquals(List.of(3, 2), keys(rangeDescending));
+ assertEquals(List.of("three", "two"), values(rangeDescending));
+ // key 3 written without headers -> empty (never null) headers
+ assertEquals(new RecordHeaders(), rangeDescending.get(0).headers());
// key 3
+ assertEquals(baseTimestamp + 1, rangeDescending.get(0).timestamp());
+ assertEquals(HEADERS, rangeDescending.get(1).headers()); //
key 2
+ assertEquals(baseTimestamp, rangeDescending.get(1).timestamp());
+ // returned headers are a read-only snapshot: no mutation (add or
remove) is allowed
+ assertThrows(IllegalStateException.class, () ->
rangeDescending.get(1).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
rangeDescending.get(1).headers().remove("source"));
+
+ // A bound matching no keys -> empty result (the no-match path).
+ final List<ReadOnlyRecord<Integer, String>> noMatch =
+ rangeQuery(TimestampedRangeWithHeadersQuery.withRange(4, 200));
+ assertTrue(noMatch.isEmpty());
+
+ // withRange(3, 2), lower > upper -> empty result (an inverted range
matches nothing, rather than failing).
+ final List<ReadOnlyRecord<Integer, String>> invertedBounds =
rangeQuery(TimestampedRangeWithHeadersQuery.withRange(3, 2));
+ assertTrue(invertedBounds.isEmpty());
+ }
+
+ @Test
+ public void shouldNotSeeUnflushedWriteInRangeQueryWhenCachingEnabled()
throws Exception {
+ // Unlike the point query (CachingKeyValueStoreWithHeaders serves
KeyQuery/TimestampedKeyWithHeadersQuery
+ // from the cache), CachingKeyValueStoreWithHeaders.query() forwards
RangeQuery/TimestampedRangeWithHeadersQuery
+ // straight to the underlying store, bypassing the cache. That
underlying store's Position is only
+ // advanced by writes that actually reach it, so a range query bound
on the input position never catches
+ // up while the write lives only in the cache -- it fails
NOT_UP_TO_BOUND, rather than returning an empty
+ // (but successful) result. Use a very large commit interval so the
record is never flushed during the test.
+ commitIntervalMs = Duration.ofMinutes(10).toMillis();
+ startStreamsWithKeyValueHeadersStore(true);
produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
- final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
+ final StateQueryRequest<ReadOnlyRecordIterator<Integer, String>>
request =
+ inStore(STORE_NAME)
+ .withQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds())
+ .withPositionBound(PositionBound.at(inputPosition));
+ final StateQueryResult<ReadOnlyRecordIterator<Integer, String>> result
= kafkaStreams.query(request);
+
+ final QueryResult<ReadOnlyRecordIterator<Integer, String>> onlyResult
= result.getOnlyPartitionResult();
+ assertTrue(onlyResult.isFailure(), "A range query bound on the input
position must not catch up while the write is cache-only");
+ assertEquals(FailureReason.NOT_UP_TO_BOUND,
onlyResult.getFailureReason());
+ }
+
+ @Test
+ public void
shouldFailWithUnknownQueryTypeForRangeQueryAgainstNonHeadersStore() throws
Exception {
+
assertUnknownQueryTypeAgainstNonHeadersStore(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds());
+ }
+
+ @Test
+ public void
shouldThrowForTimestampedRangeWithHeadersQueryOnPlainSupplier() throws
Exception {
+ // The query succeeds, but iterating throws because timestamp = -1
cannot be a ReadOnlyRecord.
+ startStreamsWithKeyValuePlainSupplierStore();
+
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
+ KeyValue.pair(1, "one"), KeyValue.pair(2, "two"));
+
+ final StateQueryRequest<ReadOnlyRecordIterator<Integer, String>>
request =
inStore(STORE_NAME)
- .withQuery(TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(1))
+ .withQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds())
.withPositionBound(PositionBound.at(inputPosition));
- final StateQueryResult<ReadOnlyRecord<Integer, String>> result =
+ final StateQueryResult<ReadOnlyRecordIterator<Integer, String>> result
=
IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
- assertTrue(result.getOnlyPartitionResult().isFailure());
- assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getOnlyPartitionResult().getFailureReason());
+ final QueryResult<ReadOnlyRecordIterator<Integer, String>> onlyResult
= result.getOnlyPartitionResult();
+ assertTrue(onlyResult.isSuccess());
+ try (ReadOnlyRecordIterator<Integer, String> iterator =
onlyResult.getResult()) {
+ assertThrows(StreamsException.class, iterator::next);
+ }
+ }
+
+ @Test
+ public void
shouldReturnEmptyHeadersForTimestampedRangeWithHeadersQueryOnAdapterStore()
throws Exception {
+ // A WithHeaders builder over a plain *timestamped* supplier
(persistentTimestampedKeyValueStore)
+ // keeps timestamps but drops headers -- distinct from the
plain-supplier build above, which can't
+ // even represent the timestamp. A range query reads the underlying
store directly, so headers
+ // come back empty (never null), even though they were written.
+ startStreamsWithKeyValueAdapterStore();
+
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "one"));
+
+ final List<ReadOnlyRecord<Integer, String>> range =
+ rangeQuery(TimestampedRangeWithHeadersQuery.<Integer,
String>withNoBounds());
+ assertEquals(List.of(1), keys(range));
+ assertEquals(List.of("one"), values(range));
+ assertEquals(baseTimestamp, range.get(0).timestamp());
+ assertEquals(new RecordHeaders(), range.get(0).headers());
+ // returned headers are a read-only snapshot: no mutation (add or
remove) is allowed, even when empty
+ assertThrows(IllegalStateException.class, () ->
range.get(0).headers().add("x", new byte[0]));
+ assertThrows(IllegalStateException.class, () ->
range.get(0).headers().remove("source"));
+ }
+
+ private void startStreams(final StoreBuilder<?> storeBuilder,
+ final ProcessorSupplier<Integer, String,
Integer, String> processorSupplier) throws Exception {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder
+ .addStateStore(storeBuilder)
+ .stream(inputStream, Consumed.with(Serdes.Integer(),
Serdes.String()))
+ .process(processorSupplier, STORE_NAME)
+ .to(outputStream, Produced.with(Serdes.Integer(),
Serdes.String()));
+
+ kafkaStreams = new KafkaStreams(builder.build(), props());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
}
- private void startStreams() throws Exception {
+ private void startStreamsWithKeyValueHeadersStore() throws Exception {
// Caching disabled: every IQv2 query is forced down to the persistent
// RocksDBTimestampedStoreWithHeaders layer, exercising its KeyQuery
handling
// (rather than being short-circuited by a cache hit).
- startStreams(false);
+ startStreamsWithKeyValueHeadersStore(false);
}
- private void startStreams(final boolean cachingEnabled) throws Exception {
- final StreamsBuilder builder = new StreamsBuilder();
+ private void startStreamsWithKeyValueHeadersStore(final boolean
cachingEnabled) throws Exception {
final StoreBuilder<TimestampedKeyValueStoreWithHeaders<Integer,
String>> storeBuilder =
Stores.timestampedKeyValueStoreWithHeadersBuilder(
Stores.persistentTimestampedKeyValueStoreWithHeaders(STORE_NAME),
Serdes.Integer(),
Serdes.String());
- builder
- .addStateStore(cachingEnabled ? storeBuilder.withCachingEnabled()
: storeBuilder.withCachingDisabled())
- .stream(inputStream, Consumed.with(Serdes.Integer(),
Serdes.String()))
- .process(() -> new HeadersStoreWriterProcessor(), STORE_NAME)
- .to(outputStream, Produced.with(Serdes.Integer(),
Serdes.String()));
+ startStreams(
+ cachingEnabled ? storeBuilder.withCachingEnabled() :
storeBuilder.withCachingDisabled(),
+ KeyValueHeadersStoreWriterProcessor::new);
+ }
- kafkaStreams = new KafkaStreams(builder.build(), props());
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+ private void startStreamsWithKeyValueNonHeadersStore() throws Exception {
+ // A plain (non-WithHeaders) timestamped store: the headers-aware
query types are unsupported here.
+ startStreams(
+ Stores.timestampedKeyValueStoreBuilder(
+ Stores.persistentTimestampedKeyValueStore(STORE_NAME),
+ Serdes.Integer(),
+ Serdes.String()),
+ KeyValuePlainStoreWriterProcessor::new);
+ }
+
+ private void startStreamsWithKeyValuePlainSupplierStore() throws Exception
{
+ // A WithHeaders builder over a plain (non-timestamped) supplier:
entries come back with
+ // timestamp = -1, which cannot be represented as a ReadOnlyRecord.
+ startStreams(
+ Stores.timestampedKeyValueStoreWithHeadersBuilder(
+ Stores.persistentKeyValueStore(STORE_NAME),
+ Serdes.Integer(),
+ Serdes.String()).withCachingDisabled(),
+ KeyValueHeadersStoreWriterProcessor::new);
+ }
+
+ private void startStreamsWithKeyValueAdapterStore() throws Exception {
+ // A WithHeaders builder over a plain *timestamped* supplier: keeps
timestamps (unlike the
+ // plain-supplier build above) but drops headers via
TimestampedToHeadersStoreAdapter.
+ startStreams(
+ Stores.timestampedKeyValueStoreWithHeadersBuilder(
+ Stores.persistentTimestampedKeyValueStore(STORE_NAME),
+ Serdes.Integer(),
+ Serdes.String()).withCachingDisabled(),
+ KeyValueHeadersStoreWriterProcessor::new);
+ }
+
+ private <R> void assertUnknownQueryTypeAgainstNonHeadersStore(final
Query<R> query) throws Exception {
+ startStreamsWithKeyValueNonHeadersStore();
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
+
+ final StateQueryRequest<R> request =
+
inStore(STORE_NAME).withQuery(query).withPositionBound(PositionBound.at(inputPosition));
+ final StateQueryResult<R> result =
IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
+
+ assertTrue(result.getOnlyPartitionResult().isFailure());
+ assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getOnlyPartitionResult().getFailureReason());
}
- private ReadOnlyRecord<Integer, String> query(final int key) {
+ private ReadOnlyRecord<Integer, String> keyQuery(final int key) {
final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
inStore(STORE_NAME)
.withQuery(TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(key))
@@ -256,7 +482,7 @@ public class IQv2HeadersStoreIntegrationTest {
return onlyResult == null ? null : onlyResult.getResult();
}
- private ReadOnlyRecord<Integer, String> querySkipCache(final int key) {
+ private ReadOnlyRecord<Integer, String> keyQuerySkipCache(final int key) {
// skipCache forwards the query past the record cache to the
persistent store. Use the default
// (unbounded) position bound on purpose: the persistent layer may
legitimately be empty (nothing
// flushed), so bounding on the input position would never be
satisfied.
@@ -268,6 +494,28 @@ public class IQv2HeadersStoreIntegrationTest {
return onlyResult == null ? null : onlyResult.getResult();
}
+ private List<ReadOnlyRecord<Integer, String>> rangeQuery(final
TimestampedRangeWithHeadersQuery<Integer, String> query) {
+ final StateQueryRequest<ReadOnlyRecordIterator<Integer, String>>
request =
+
inStore(STORE_NAME).withQuery(query).withPositionBound(PositionBound.at(inputPosition));
+ final StateQueryResult<ReadOnlyRecordIterator<Integer, String>> result
=
+ IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
+ final List<ReadOnlyRecord<Integer, String>> records = new
ArrayList<>();
+ try (ReadOnlyRecordIterator<Integer, String> iterator =
result.getOnlyPartitionResult().getResult()) {
+ while (iterator.hasNext()) {
+ records.add(iterator.next());
+ }
+ }
+ return records;
+ }
+
+ private static List<Integer> keys(final List<ReadOnlyRecord<Integer,
String>> records) {
+ return
records.stream().map(ReadOnlyRecord::key).collect(Collectors.toList());
+ }
+
+ private static List<String> values(final List<ReadOnlyRecord<Integer,
String>> records) {
+ return
records.stream().map(ReadOnlyRecord::value).collect(Collectors.toList());
+ }
+
private Properties props() {
final String safeTestName = safeUniqueTestName(testInfo);
final Properties streamsConfiguration = new Properties();
@@ -307,7 +555,7 @@ public class IQv2HeadersStoreIntegrationTest {
}
}
- private static class HeadersStoreWriterProcessor implements
Processor<Integer, String, Integer, String> {
+ private static class KeyValueHeadersStoreWriterProcessor implements
Processor<Integer, String, Integer, String> {
private ProcessorContext<Integer, String> context;
private TimestampedKeyValueStoreWithHeaders<Integer, String> store;
@@ -319,14 +567,18 @@ public class IQv2HeadersStoreIntegrationTest {
@Override
public void process(final Record<Integer, String> record) {
- store.put(
- record.key(),
- ValueTimestampHeaders.make(record.value(), record.timestamp(),
record.headers()));
+ if (record.value() == null) {
+ store.delete(record.key());
+ } else {
+ store.put(
+ record.key(),
+ ValueTimestampHeaders.make(record.value(),
record.timestamp(), record.headers()));
+ }
context.forward(record);
}
}
- private static class PlainStoreWriterProcessor implements
Processor<Integer, String, Integer, String> {
+ private static class KeyValuePlainStoreWriterProcessor implements
Processor<Integer, String, Integer, String> {
private ProcessorContext<Integer, String> context;
private TimestampedKeyValueStore<Integer, String> store;
diff --git
a/streams/src/main/java/org/apache/kafka/streams/query/TimestampedRangeWithHeadersQuery.java
b/streams/src/main/java/org/apache/kafka/streams/query/TimestampedRangeWithHeadersQuery.java
new file mode 100644
index 00000000000..65762c72f57
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/query/TimestampedRangeWithHeadersQuery.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.query;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+import org.apache.kafka.common.annotation.InterfaceStability.Evolving;
+import org.apache.kafka.streams.processor.api.ReadOnlyRecord;
+import org.apache.kafka.streams.state.ReadOnlyRecordIterator;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+
+import java.util.Optional;
+
+/**
+ * Interactive query for issuing range queries and scans over a
+ * {@link TimestampedKeyValueStoreWithHeaders}, returning each record together
with its headers.
+ *
+ * <p>This is the headers-aware parallel of {@link TimestampedRangeQuery}: it
returns a
+ * {@link ReadOnlyRecordIterator} of {@link ReadOnlyRecord} elements, each
carrying the key, value,
+ * timestamp, and headers, whereas {@link TimestampedRangeQuery} returns a
+ * {@link org.apache.kafka.streams.state.KeyValueIterator} of
+ * {@link org.apache.kafka.streams.state.ValueAndTimestamp} (value and
timestamp only, no headers).
+ *
+ * <p>A range query retrieves a set of records, specified using an upper
and/or lower bound on the
+ * keys. A scan query (no bounds) retrieves all records contained in the
store. Keys' order is based
+ * on the serialized {@code byte[]} of the keys, not the 'logical' key order.
+ *
+ * <p>Headers are persisted and returned only when the store is backed by a
native headers store,
+ * i.e. built with a KIP-1271 {@code WithHeaders} byte-store supplier (e.g.
+ * {@code Stores.persistentTimestampedKeyValueStoreWithHeaders}). A {@code
WithHeaders} store built
+ * over a legacy (non-headers) supplier cannot persist headers, so the
store-served reads come back
+ * with empty {@code headers()}.
+ *
+ * <p>Against a plain store not built with the {@code WithHeaders} builder at
all, this query type is
+ * unsupported and fails with {@link FailureReason#UNKNOWN_QUERY_TYPE}.
+ *
+ * <p>Each element is a {@link ReadOnlyRecord}, whose timestamp is
non-negative. If the backing store
+ * does not persist timestamps -- for example a {@code WithHeaders} store
built over a plain
+ * {@link org.apache.kafka.streams.state.KeyValueStore} supplier, which
surfaces every entry with
+ * {@code NO_TIMESTAMP} ({@code -1}) -- that entry cannot be represented, so
advancing the returned
+ * {@link ReadOnlyRecordIterator} throws {@link
org.apache.kafka.streams.errors.StreamsException} at
+ * that entry. Back the store with {@code
Stores.persistentTimestampedKeyValueStoreWithHeaders(...)} to
+ * persist timestamps and headers, or use {@link TimestampedRangeQuery} if
headers are not needed.
+ *
+ * @param <K> Type of keys
+ * @param <V> Type of values
+ */
+@Evolving
[email protected]
+public final class TimestampedRangeWithHeadersQuery<K, V> implements
Query<ReadOnlyRecordIterator<K, V>> {
+
+ private final Optional<K> lower;
+ private final Optional<K> upper;
+ private final ResultOrder order;
+
+ private TimestampedRangeWithHeadersQuery(final Optional<K> lower, final
Optional<K> upper, final ResultOrder order) {
+ this.lower = lower;
+ this.upper = upper;
+ this.order = order;
+ }
+
+ /**
+ * Interactive range query using a lower and upper bound to filter the
keys returned.
+ * @param lower The key that specifies the lower bound of the range
+ * @param upper The key that specifies the upper bound of the range
+ * @param <K> The key type
+ * @param <V> The value type
+ */
+ public static <K, V> TimestampedRangeWithHeadersQuery<K, V>
withRange(final K lower, final K upper) {
+ return new
TimestampedRangeWithHeadersQuery<>(Optional.ofNullable(lower),
Optional.ofNullable(upper), ResultOrder.ANY);
+ }
+
+ /**
+ * Interactive range query using an upper bound to filter the keys
returned.
+ * @param upper The key that specifies the upper bound of the range
+ * @param <K> The key type
+ * @param <V> The value type
+ */
+ public static <K, V> TimestampedRangeWithHeadersQuery<K, V>
withUpperBound(final K upper) {
+ return new TimestampedRangeWithHeadersQuery<>(Optional.empty(),
Optional.of(upper), ResultOrder.ANY);
+ }
+
+ /**
+ * Interactive range query using a lower bound to filter the keys returned.
+ * @param lower The key that specifies the lower bound of the range
+ * @param <K> The key type
+ * @param <V> The value type
+ */
+ public static <K, V> TimestampedRangeWithHeadersQuery<K, V>
withLowerBound(final K lower) {
+ return new TimestampedRangeWithHeadersQuery<>(Optional.of(lower),
Optional.empty(), ResultOrder.ANY);
+ }
+
+ /**
+ * Interactive scan query that returns all records in the store.
+ * @param <K> The key type
+ * @param <V> The value type
+ */
+ public static <K, V> TimestampedRangeWithHeadersQuery<K, V> withNoBounds()
{
+ return new TimestampedRangeWithHeadersQuery<>(Optional.empty(),
Optional.empty(), ResultOrder.ANY);
+ }
+
+ /**
+ * Set the query to return the serialized {@code byte[]} of the keys in
descending order.
+ * Order is based on the serialized {@code byte[]} of the keys, not the
'logical' key order.
+ * @return a new query instance with the descending flag set.
+ */
+ public TimestampedRangeWithHeadersQuery<K, V> withDescendingKeys() {
+ return new TimestampedRangeWithHeadersQuery<>(this.lower, this.upper,
ResultOrder.DESCENDING);
+ }
+
+ /**
+ * Set the query to return the serialized {@code byte[]} of the keys in
ascending order.
+ * Order is based on the serialized {@code byte[]} of the keys, not the
'logical' key order.
+ * @return a new query instance with the ascending flag set.
+ */
+ public TimestampedRangeWithHeadersQuery<K, V> withAscendingKeys() {
+ return new TimestampedRangeWithHeadersQuery<>(this.lower, this.upper,
ResultOrder.ASCENDING);
+ }
+
+ /**
+ * The lower bound of the query, if specified.
+ */
+ public Optional<K> lowerBound() {
+ return lower;
+ }
+
+ /**
+ * The upper bound of the query, if specified.
+ */
+ public Optional<K> upperBound() {
+ return upper;
+ }
+
+ /**
+ * Determines if the serialized {@code byte[]} of the keys in ascending or
descending or unordered order.
+ * Order is based on the serialized {@code byte[]} of the keys, not the
'logical' key order.
+ * @return the order of the returned records based on the serialized
{@code byte[]} of the keys
+ * (can be unordered, ascending, or descending).
+ */
+ public ResultOrder resultOrder() {
+ return order;
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java
index 98b6a024b22..3d10e5569cc 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java
@@ -26,6 +26,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.ProcessorStateException;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.api.ReadOnlyRecord;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
@@ -41,10 +42,12 @@ import org.apache.kafka.streams.query.ResultOrder;
import org.apache.kafka.streams.query.TimestampedKeyQuery;
import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
import org.apache.kafka.streams.query.TimestampedRangeQuery;
+import org.apache.kafka.streams.query.TimestampedRangeWithHeadersQuery;
import org.apache.kafka.streams.query.internals.InternalQueryResultUtil;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyRecordIterator;
import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
@@ -54,6 +57,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.function.Function;
import static org.apache.kafka.common.utils.Utils.mkEntry;
@@ -106,6 +110,10 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
mkEntry(
TimestampedRangeQuery.class,
(query, positionBound, config, store) ->
runTimestampedRangeQuery(query, positionBound, config)
+ ),
+ mkEntry(
+ TimestampedRangeWithHeadersQuery.class,
+ (query, positionBound, config, store) ->
runTimestampedRangeWithHeadersQuery(query, positionBound, config)
)
);
@@ -451,6 +459,22 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
return result;
}
+ private RangeQuery<Bytes, byte[]> rawRangeQuery(final Optional<K>
lowerBound,
+ final Optional<K>
upperBound,
+ final ResultOrder order) {
+ RangeQuery<Bytes, byte[]> rawRangeQuery = RangeQuery.withRange(
+ serializeKey(lowerBound.orElse(null), internalContext.headers()),
+ serializeKey(upperBound.orElse(null), internalContext.headers())
+ );
+ if (order.equals(ResultOrder.DESCENDING)) {
+ rawRangeQuery = rawRangeQuery.withDescendingKeys();
+ }
+ if (order.equals(ResultOrder.ASCENDING)) {
+ rawRangeQuery = rawRangeQuery.withAscendingKeys();
+ }
+ return rawRangeQuery;
+ }
+
@SuppressWarnings("unchecked")
private <R> QueryResult<R> runRangeQuery(
final Query<R> query,
@@ -460,18 +484,8 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
final QueryResult<R> result;
final RangeQuery<K, V> typedQuery = (RangeQuery<K, V>) query;
- RangeQuery<Bytes, byte[]> rawRangeQuery;
- final ResultOrder order = typedQuery.resultOrder();
- rawRangeQuery = RangeQuery.withRange(
- serializeKey(typedQuery.getLowerBound().orElse(null),
internalContext.headers()),
- serializeKey(typedQuery.getUpperBound().orElse(null),
internalContext.headers())
- );
- if (order.equals(ResultOrder.DESCENDING)) {
- rawRangeQuery = rawRangeQuery.withDescendingKeys();
- }
- if (order.equals(ResultOrder.ASCENDING)) {
- rawRangeQuery = rawRangeQuery.withAscendingKeys();
- }
+ final RangeQuery<Bytes, byte[]> rawRangeQuery =
+ rawRangeQuery(typedQuery.getLowerBound(),
typedQuery.getUpperBound(), typedQuery.resultOrder());
final QueryResult<KeyValueIterator<Bytes, byte[]>> rawResult =
wrapped().query(rawRangeQuery, positionBound, config);
if (rawResult.isSuccess()) {
@@ -505,18 +519,8 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
final QueryResult<R> result;
final TimestampedRangeQuery<K, V> typedQuery =
(TimestampedRangeQuery<K, V>) query;
- RangeQuery<Bytes, byte[]> rawRangeQuery;
- final ResultOrder order = typedQuery.resultOrder();
- rawRangeQuery = RangeQuery.withRange(
- serializeKey(typedQuery.lowerBound().orElse(null),
internalContext.headers()),
- serializeKey(typedQuery.upperBound().orElse(null),
internalContext.headers())
- );
- if (order.equals(ResultOrder.DESCENDING)) {
- rawRangeQuery = rawRangeQuery.withDescendingKeys();
- }
- if (order.equals(ResultOrder.ASCENDING)) {
- rawRangeQuery = rawRangeQuery.withAscendingKeys();
- }
+ final RangeQuery<Bytes, byte[]> rawRangeQuery =
+ rawRangeQuery(typedQuery.lowerBound(), typedQuery.upperBound(),
typedQuery.resultOrder());
final QueryResult<KeyValueIterator<Bytes, byte[]>> rawResult =
wrapped().query(rawRangeQuery, positionBound, config);
if (rawResult.isSuccess()) {
@@ -542,6 +546,40 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
return result;
}
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runTimestampedRangeWithHeadersQuery(
+ final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config
+ ) {
+ final QueryResult<R> result;
+ final TimestampedRangeWithHeadersQuery<K, V> typedQuery =
(TimestampedRangeWithHeadersQuery<K, V>) query;
+
+ final RangeQuery<Bytes, byte[]> rawRangeQuery =
+ rawRangeQuery(typedQuery.lowerBound(), typedQuery.upperBound(),
typedQuery.resultOrder());
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> rawResult =
wrapped().query(rawRangeQuery, positionBound, config);
+ if (rawResult.isSuccess()) {
+ final KeyValueIterator<Bytes, byte[]> iterator =
rawResult.getResult();
+ final ReadOnlyRecordIterator<K, V> resultIterator =
+ new
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator(
+ iterator,
+ getSensor,
+ StoreQueryUtils.deserializeValue(serdes, wrapped())
+ );
+ final QueryResult<ReadOnlyRecordIterator<K, V>> typedQueryResult =
+ InternalQueryResultUtil.copyAndSubstituteDeserializedResult(
+ rawResult,
+ resultIterator
+ );
+ result = (QueryResult<R>) typedQueryResult;
+ } else {
+ // the generic type doesn't matter, since failed queries have no
result set.
+ result = (QueryResult<R>) rawResult;
+ }
+ return result;
+ }
+
@Override
public <PS extends Serializer<P>, P> KeyValueIterator<K,
ValueTimestampHeaders<V>> prefixScan(
final P prefix, final PS prefixKeySerializer
@@ -615,13 +653,49 @@ public class
MeteredTimestampedKeyValueStoreWithHeaders<K, V>
return new
MeteredTimestampedKeyValueStoreWithHeadersIterator(store.reverseAll(),
allSensor);
}
- @SuppressWarnings("unchecked")
- private class MeteredTimestampedKeyValueStoreWithHeadersQueryIterator
implements KeyValueIterator<K, V>, MeteredIterator {
+ /**
+ * Shared scaffolding for the metered iterators below: tracks {@code
num-open-iterators},
+ * {@code oldest-iterator-open-since-ms}, and per-operation iterator
duration, and delegates
+ * closing the wrapped raw iterator. Subclasses only need to implement the
deserializing
+ * {@code next()}/{@code hasNext()} (and, where applicable, {@code
peekNextKey()}).
+ */
+ private abstract class AbstractMeteredIterator implements MeteredIterator {
- private final KeyValueIterator<Bytes, byte[]> iter;
+ final KeyValueIterator<Bytes, byte[]> iter;
private final Sensor sensor;
private final long startNs;
private final long startTimestampMs;
+
+ AbstractMeteredIterator(final KeyValueIterator<Bytes, byte[]> iter,
final Sensor sensor) {
+ this.iter = iter;
+ this.sensor = sensor;
+ this.startNs = time.nanoseconds();
+ this.startTimestampMs = time.milliseconds();
+ numOpenIterators.increment();
+ openIterators.add(this);
+ }
+
+ @Override
+ public long startTimestamp() {
+ return startTimestampMs;
+ }
+
+ public void close() {
+ try {
+ iter.close();
+ } finally {
+ final long duration = time.nanoseconds() - startNs;
+ sensor.record(duration);
+ iteratorDurationSensor.record(duration);
+ numOpenIterators.decrement();
+ openIterators.remove(this);
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private class MeteredTimestampedKeyValueStoreWithHeadersQueryIterator
extends AbstractMeteredIterator implements KeyValueIterator<K, V> {
+
private final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer;
private final boolean returnPlainValue;
@@ -633,18 +707,9 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer,
final boolean returnPlainValue
) {
- this.iter = iter;
- this.sensor = sensor;
+ super(iter, sensor);
this.valueTimestampHeadersDeserializer =
valueTimestampHeadersDeserializer;
- this.startNs = time.nanoseconds();
- this.startTimestampMs = time.milliseconds();
this.returnPlainValue = returnPlainValue;
- openIterators.add(this);
- }
-
- @Override
- public long startTimestamp() {
- return startTimestampMs;
}
@Override
@@ -675,18 +740,6 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
}
}
- @Override
- public void close() {
- try {
- iter.close();
- } finally {
- final long duration = time.nanoseconds() - startNs;
- sensor.record(duration);
- iteratorDurationSensor.record(duration);
- openIterators.remove(this);
- }
- }
-
@Override
public K peekNextKey() {
if (cachedNext == null) {
@@ -696,28 +749,79 @@ public class
MeteredTimestampedKeyValueStoreWithHeaders<K, V>
}
}
- private class MeteredTimestampedKeyValueStoreWithHeadersIterator
implements KeyValueIterator<K, ValueTimestampHeaders<V>>, MeteredIterator {
- private final KeyValueIterator<Bytes, byte[]> iter;
- private final Sensor sensor;
- private final long startNs;
- private final long startTimestampMs;
- private KeyValue<K, ValueTimestampHeaders<V>> cachedNext;
+ /**
+ * Iterator backing {@link TimestampedRangeWithHeadersQuery}: yields each
entry as a
+ * {@link ReadOnlyRecord} (implemented by {@link Record}) carrying key,
value, timestamp, and the
+ * stored headers, with the headers frozen so a caller cannot mutate the
read-only result.
+ *
+ * <p>A {@link ReadOnlyRecord} timestamp is contractually non-negative, so
an entry with a negative
+ * stored timestamp cannot be represented. The dominant, deterministic
cause is a store that does
+ * not persist timestamps: a {@code WithHeaders} store built over a plain
{@link KeyValueStore}
+ * supplier surfaces every entry with {@code NO_TIMESTAMP} (-1). A genuine
negative write is
+ * otherwise blocked (the source {@code RecordQueue} drops
negative-timestamp records at ingestion),
+ * though {@link ValueAndTimestamp#make}/{@link
ValueTimestampHeaders#make} do not themselves reject
+ * one written directly.
+ *
+ * <p>This mirrors the rule the point query {@link
TimestampedKeyWithHeadersQuery} applies. But
+ * because a lazily-evaluated range has already returned a successful
{@link QueryResult} before any
+ * entry is read, such an entry cannot be surfaced as a query-level
failure; it is instead reported
+ * by throwing a {@link StreamsException} while advancing the iterator.
+ *
+ * <p>That throw does not close the iterator: a caller that catches it and
abandons the iterator
+ * leaks the underlying raw iterator and permanently inflates {@code
num-open-iterators}. Callers
+ * must close this iterator in a {@code finally} block or a
try-with-resources statement, even when
+ * {@code next()} throws.
+ */
+ private class
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator
+ extends AbstractMeteredIterator implements ReadOnlyRecordIterator<K,
V> {
- private MeteredTimestampedKeyValueStoreWithHeadersIterator(
+ private final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer;
+
+ private
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator(
final KeyValueIterator<Bytes, byte[]> iter,
- final Sensor sensor
+ final Sensor sensor,
+ final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer
) {
- this.iter = iter;
- this.sensor = sensor;
- this.startNs = time.nanoseconds();
- this.startTimestampMs = time.milliseconds();
- numOpenIterators.increment();
- openIterators.add(this);
+ super(iter, sensor);
+ this.valueTimestampHeadersDeserializer =
valueTimestampHeadersDeserializer;
}
@Override
- public long startTimestamp() {
- return startTimestampMs;
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public ReadOnlyRecord<K, V> next() {
+ final KeyValue<Bytes, byte[]> keyValue = iter.next();
+ final ValueTimestampHeaders<V> valueTimestampHeaders =
valueTimestampHeadersDeserializer.apply(keyValue.value);
+ final Headers headers = valueTimestampHeaders.headers();
+ final K key = deserializeKey(keyValue.key.get(), headers);
+ if (valueTimestampHeaders.timestamp() < 0) {
+ throw new StreamsException(
+ "Cannot represent the stored record for key [" + key + "]
as a ReadOnlyRecord: its "
+ + "timestamp (" + valueTimestampHeaders.timestamp() +
") is negative.");
+ }
+ final Record<K, V> record = new Record<>(
+ key,
+ valueTimestampHeaders.value(),
+ valueTimestampHeaders.timestamp(),
+ headers);
+ ((RecordHeaders) record.headers()).setReadOnly();
+ return record;
+ }
+ }
+
+ private class MeteredTimestampedKeyValueStoreWithHeadersIterator
+ extends AbstractMeteredIterator implements KeyValueIterator<K,
ValueTimestampHeaders<V>> {
+
+ private KeyValue<K, ValueTimestampHeaders<V>> cachedNext;
+
+ private MeteredTimestampedKeyValueStoreWithHeadersIterator(
+ final KeyValueIterator<Bytes, byte[]> iter,
+ final Sensor sensor
+ ) {
+ super(iter, sensor);
}
@Override
@@ -739,19 +843,6 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
return KeyValue.pair(key, valueTimestampHeaders);
}
- @Override
- public void close() {
- try {
- iter.close();
- } finally {
- final long duration = time.nanoseconds() - startNs;
- sensor.record(duration);
- iteratorDurationSensor.record(duration);
- numOpenIterators.decrement();
- openIterators.remove(this);
- }
- }
-
@Override
public K peekNextKey() {
if (cachedNext == null) {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
index d0ad49acfcd..931889c58d2 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeaders.java
@@ -18,11 +18,6 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.streams.errors.ProcessorStateException;
-import org.apache.kafka.streams.query.KeyQuery;
-import org.apache.kafka.streams.query.PositionBound;
-import org.apache.kafka.streams.query.Query;
-import org.apache.kafka.streams.query.QueryConfig;
-import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.state.HeadersBytesStore;
import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecorder;
@@ -195,33 +190,4 @@ public class RocksDBTimestampedStoreWithHeaders extends
RocksDBStore implements
}
}
- @SuppressWarnings("SynchronizeOnNonFinalField")
- @Override
- public <R> QueryResult<R> query(final Query<R> query,
- final PositionBound positionBound,
- final QueryConfig config) {
- // KIP-1356 (point query only): the headers-aware
TimestampedKeyWithHeadersQuery forwards a raw
- // KeyQuery to this native store, so enable KeyQuery via the inherited
RocksDBStore handling.
- // Every other query type (e.g. RangeQuery) stays UNKNOWN_QUERY_TYPE
here, unchanged from before;
- // those are enabled by their own KIP-1356 follow-ups.
- if (query instanceof KeyQuery) {
- return super.query(query, positionBound, config);
- }
-
- final long start = config.isCollectExecutionInfo() ? System.nanoTime()
: -1L;
- final QueryResult<R> result;
-
- synchronized (position) {
- result = QueryResult.forUnknownQueryType(query, this);
-
- if (config.isCollectExecutionInfo()) {
- result.addExecutionInfo(
- "Handled in " + this.getClass() + " in " +
(System.nanoTime() - start) + "ns"
- );
- }
- result.setPosition(position.copy());
- }
- return result;
- }
-
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java
index 568a11a7226..7a45afae1e8 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java
@@ -36,14 +36,20 @@ import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.ProcessorStateManager;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.query.FailureReason;
import org.apache.kafka.streams.query.KeyQuery;
import org.apache.kafka.streams.query.PositionBound;
import org.apache.kafka.streams.query.Query;
import org.apache.kafka.streams.query.QueryConfig;
import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.streams.query.RangeQuery;
+import org.apache.kafka.streams.query.ResultOrder;
import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
+import org.apache.kafka.streams.query.TimestampedRangeQuery;
+import org.apache.kafka.streams.query.TimestampedRangeWithHeadersQuery;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyRecordIterator;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
import org.apache.kafka.test.KeyValueIteratorStub;
@@ -55,6 +61,8 @@ import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -65,6 +73,7 @@ import static org.apache.kafka.common.utils.Utils.mkMap;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
@@ -89,7 +98,9 @@ public class MeteredTimestampedKeyValueStoreWithHeadersTest {
private static final RecordHeaders HEADERS = makeHeaders();
private static final ValueTimestampHeaders<String> VALUE_TIMESTAMP_HEADERS
=
ValueTimestampHeaders.make("value", 97L, HEADERS);
- private static final byte[] VALUE_TIMESTAMP_HEADERS_BYTES =
serializeValueTimestampHeaders();
+ private static final byte[] VALUE_TIMESTAMP_HEADERS_BYTES =
serializeValueTimestampHeaders(VALUE_TIMESTAMP_HEADERS);
+ private static final byte[]
NEGATIVE_TIMESTAMP_VALUE_TIMESTAMP_HEADERS_BYTES =
+ serializeValueTimestampHeaders(ValueTimestampHeaders.make("value",
-1L, HEADERS));
private final String threadId = Thread.currentThread().getName();
private final TaskId taskId = new TaskId(0, 0, "My-Topology");
@Mock
@@ -302,6 +313,166 @@ public class
MeteredTimestampedKeyValueStoreWithHeadersTest {
assertTrue((Double) metric.metricValue() > 0);
}
+ @Test
+ public void
shouldPropagateLowerBoundAndDescendingOrderForTimestampedRangeWithHeadersQuery()
{
+ setUp();
+ init();
+ final RangeQuery<?, ?> rawQuery = forwardedRawRangeQuery(
+ TimestampedRangeWithHeadersQuery.<String,
String>withLowerBound("a").withDescendingKeys());
+ assertEquals(ResultOrder.DESCENDING, rawQuery.resultOrder());
+ assertEquals(Bytes.wrap("a".getBytes()),
rawQuery.getLowerBound().get());
+ assertFalse(rawQuery.getUpperBound().isPresent());
+ }
+
+ @Test
+ public void
shouldPropagateUpperBoundAndAscendingOrderForTimestampedRangeWithHeadersQuery()
{
+ setUp();
+ init();
+ final RangeQuery<?, ?> rawQuery = forwardedRawRangeQuery(
+ TimestampedRangeWithHeadersQuery.<String,
String>withUpperBound("z").withAscendingKeys());
+ assertEquals(ResultOrder.ASCENDING, rawQuery.resultOrder());
+ assertFalse(rawQuery.getLowerBound().isPresent());
+ assertEquals(Bytes.wrap("z".getBytes()),
rawQuery.getUpperBound().get());
+ }
+
+ @Test
+ public void
shouldPropagateNoBoundsAndAnyOrderForTimestampedRangeWithHeadersQuery() {
+ setUp();
+ init();
+ final RangeQuery<?, ?> rawQuery = forwardedRawRangeQuery(
+ TimestampedRangeWithHeadersQuery.withNoBounds());
+ assertEquals(ResultOrder.ANY, rawQuery.resultOrder());
+ assertFalse(rawQuery.getLowerBound().isPresent());
+ assertFalse(rawQuery.getUpperBound().isPresent());
+ }
+
+ @Test
+ public void
shouldPropagateBothBoundsAndAnyOrderForTimestampedRangeWithHeadersQuery() {
+ setUp();
+ init();
+ final RangeQuery<?, ?> rawQuery = forwardedRawRangeQuery(
+ TimestampedRangeWithHeadersQuery.withRange("a", "z"));
+ assertEquals(ResultOrder.ANY, rawQuery.resultOrder());
+ assertEquals(Bytes.wrap("a".getBytes()),
rawQuery.getLowerBound().get());
+ assertEquals(Bytes.wrap("z".getBytes()),
rawQuery.getUpperBound().get());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void shouldPropagateFailureForTimestampedRangeWithHeadersQuery() {
+ setUp();
+ init();
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> rawFailure =
+ QueryResult.forFailure(FailureReason.NOT_UP_TO_BOUND, "not yet
caught up");
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class))).thenReturn((QueryResult) rawFailure);
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
metered.query(
+ TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertFalse(result.isSuccess());
+ assertEquals(FailureReason.NOT_UP_TO_BOUND, result.getFailureReason());
+ }
+
+ // The num-open-iterators gauge is tracked by every metered iterator this
store returns from query().
+ // Each query type is served by a different iterator class, so each needs
its own 0->1->0 guard:
+ // - TimestampedRangeWithHeadersQuery ->
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator
+ // - RangeQuery / TimestampedRangeQuery ->
MeteredTimestampedKeyValueStoreWithHeadersQueryIterator
+ // (the all()/range() KeyValueIterator path is covered separately by
shouldTrackOpenIteratorsMetric).
+
+ @Test
+ public void
shouldTrackOpenIteratorsMetricForTimestampedRangeWithHeadersQuery() throws
IOException {
+
assertQueryTracksOpenIteratorsMetric(TimestampedRangeWithHeadersQuery.withNoBounds());
+ }
+
+ @Test
+ public void shouldTrackOpenIteratorsMetricForRangeQuery() throws
IOException {
+ assertQueryTracksOpenIteratorsMetric(RangeQuery.withNoBounds());
+ }
+
+ @Test
+ public void shouldTrackOpenIteratorsMetricForTimestampedRangeQuery()
throws IOException {
+
assertQueryTracksOpenIteratorsMetric(TimestampedRangeQuery.withNoBounds());
+ }
+
+ @SuppressWarnings("unchecked")
+ private void assertQueryTracksOpenIteratorsMetric(final Query<?> query)
throws IOException {
+ setUp();
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn((QueryResult)
QueryResult.forResult(KeyValueIterators.emptyIterator()));
+ init();
+
+ final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
+ assertEquals(0L, (Long) openIteratorsMetric.metricValue());
+
+ // The result iterator type differs per query (ReadOnlyRecordIterator
vs KeyValueIterator); both are
+ // Closeable, which is all this gauge check needs.
+ try (Closeable iterator = (Closeable) metered.query(
+ query, PositionBound.unbounded(), new
QueryConfig(false)).getResult()) {
+ assertEquals(1L, (Long) openIteratorsMetric.metricValue());
+ }
+
+ assertEquals(0L, (Long) openIteratorsMetric.metricValue());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void
shouldDecrementOpenIteratorsTwiceWhenClosedTwiceForTimestampedRangeWithHeadersQuery()
{
+ // Documents current behavior: close() is not idempotent, so a caller
that (incorrectly) closes
+ // twice double-decrements the gauge. This is a misuse path, not a
supported usage pattern.
+ setUp();
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn((QueryResult)
QueryResult.forResult(KeyValueIterators.emptyIterator()));
+ init();
+
+ final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
metered.query(
+ TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+ final ReadOnlyRecordIterator<String, String> iterator =
result.getResult();
+
+ iterator.close();
+ assertEquals(0L, (Long) openIteratorsMetric.metricValue());
+ iterator.close();
+ assertEquals(-1L, (Long) openIteratorsMetric.metricValue(), "A second
close() double-decrements the gauge");
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test
+ public void
shouldLeaveIteratorOpenWhenNextThrowsAndNotClosedForTimestampedRangeWithHeadersQuery()
{
+ // Documents current behavior: an entry with a negative stored
timestamp makes next() throw
+ // instead of returning, and that throw does not close the iterator. A
caller that catches the
+ // exception without closing (in violation of the class's contract)
leaks the open-iterator count.
+ setUp();
+ final KeyValue<Bytes, byte[]> negativeTimestampEntry =
+ KeyValue.pair(KEY_BYTES,
NEGATIVE_TIMESTAMP_VALUE_TIMESTAMP_HEADERS_BYTES);
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn((QueryResult) QueryResult.forResult(
+ new
KeyValueIteratorStub<>(List.of(negativeTimestampEntry).iterator())));
+ init();
+
+ final KafkaMetric openIteratorsMetric = metric("num-open-iterators");
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
metered.query(
+ TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+ final ReadOnlyRecordIterator<String, String> iterator =
result.getResult();
+
+ assertThrows(StreamsException.class, iterator::next);
+ assertEquals(1L, (Long) openIteratorsMetric.metricValue(),
+ "An uncaught/unclosed iterator after next() throws must remain
open");
+
+ iterator.close();
+ assertEquals(0L, (Long) openIteratorsMetric.metricValue());
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private RangeQuery<?, ?> forwardedRawRangeQuery(final Query<?> query) {
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn((QueryResult) QueryResult.forResult(
+ new KeyValueIteratorStub<>(List.<KeyValue<Bytes,
byte[]>>of().iterator())));
+ metered.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+ final ArgumentCaptor<RangeQuery> captor =
ArgumentCaptor.forClass(RangeQuery.class);
+ verify(inner).query(captor.capture(), any(PositionBound.class),
any(QueryConfig.class));
+ return captor.getValue();
+ }
+
@Test
public void shouldGetAllFromInnerStoreAndRecordAllMetric() {
setUp();
@@ -495,9 +666,9 @@ public class MeteredTimestampedKeyValueStoreWithHeadersTest
{
return headers;
}
- private static byte[] serializeValueTimestampHeaders() {
+ private static byte[] serializeValueTimestampHeaders(final
ValueTimestampHeaders<String> valueTimestampHeaders) {
final ValueTimestampHeadersSerializer<String> serializer = new
ValueTimestampHeadersSerializer<>(Serdes.String().serializer());
- return serializer.serialize("topic", VALUE_TIMESTAMP_HEADERS);
+ return serializer.serialize("topic", valueTimestampHeaders);
}
@SuppressWarnings("unchecked")
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
index be048ece610..c1b1259d7f8 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStoreWithHeadersTest.java
@@ -1030,23 +1030,30 @@ public class RocksDBTimestampedStoreWithHeadersTest
extends RocksDBStoreTest {
}
@Test
- public void shouldReturnUnknownQueryTypeForRangeQuery() {
+ public void shouldHandleRangeQuery() {
// Initialize the store
rocksDBStore.init(context, rocksDBStore);
- // KIP-1356 (point query only): this store enables KeyQuery but leaves
RangeQuery (and every
- // other non-KeyQuery type) as UNKNOWN_QUERY_TYPE, unchanged from
before. RangeQuery support on
- // the native header store is deferred to the range KIP-1356 follow-up.
+ // Store raw (serialized ValueTimestampHeaders) bytes for a key.
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] storedBytes = "headers+timestamp+value".getBytes();
+ rocksDBStore.put(key, storedBytes);
+
+ // KIP-1356: the range follow-up enables RangeQuery on the native
header store via the inherited
+ // RocksDBStore handling (returning the raw stored header-format
bytes; the metered store does
+ // the header-aware deserialization), matching the adapter build path.
final RangeQuery<Bytes, byte[]> query = RangeQuery.withNoBounds();
final QueryResult<KeyValueIterator<Bytes, byte[]>> result =
rocksDBStore.query(query, PositionBound.unbounded(), new
QueryConfig(false));
- assertFalse(result.isSuccess(), "Expected RangeQuery to be
unsupported");
- assertEquals(
- FailureReason.UNKNOWN_QUERY_TYPE,
- result.getFailureReason(),
- "Expected UNKNOWN_QUERY_TYPE failure reason"
- );
+ assertTrue(result.isSuccess(), "Expected RangeQuery to succeed");
+ try (KeyValueIterator<Bytes, byte[]> iterator = result.getResult()) {
+ assertTrue(iterator.hasNext(), "Expected the stored key in the
range result");
+ final KeyValue<Bytes, byte[]> keyValue = iterator.next();
+ assertEquals(key, keyValue.key);
+ assertArrayEquals(storedBytes, keyValue.value, "Expected the raw
stored bytes to be returned");
+ assertFalse(iterator.hasNext());
+ }
assertNotNull(result.getPosition(), "Expected position to be set");
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java
index cccc1174e53..1863e977b23 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java
@@ -24,6 +24,8 @@ import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.api.ReadOnlyRecord;
import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
@@ -37,8 +39,12 @@ import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.query.RangeQuery;
import org.apache.kafka.streams.query.TimestampedKeyQuery;
import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
+import org.apache.kafka.streams.query.TimestampedRangeQuery;
+import org.apache.kafka.streams.query.TimestampedRangeWithHeadersQuery;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyRecordIterator;
import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
@@ -56,7 +62,9 @@ import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -481,24 +489,6 @@ public class
TimestampedKeyValueStoreBuilderWithHeadersTest {
}
}
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void shouldReturnUnknownQueryTypeForRangeQueryOnHeadersStore(final
boolean cachingEnabled) {
- // KIP-1356 (point query only): the native header store enables
KeyQuery but not RangeQuery, so a
- // RangeQuery reports UNKNOWN_QUERY_TYPE (caching on or off).
RangeQuery support on the native
- // build is deferred to the range follow-up.
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, cachingEnabled);
- try {
- final QueryResult<?> result =
- store.query(RangeQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
- assertFalse(result.isSuccess(), "Expected RangeQuery to be
unsupported on the native header store");
- assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getFailureReason());
- assertNotNull(result.getPosition(), "Expected position to be set");
- } finally {
- store.close();
- }
- }
-
@ParameterizedTest
@CsvSource({"NATIVE", "ADAPTER", "IN_MEMORY"})
public void shouldReturnEmptyPositionInitially(final StoreType storeType) {
@@ -615,4 +605,229 @@ public class
TimestampedKeyValueStoreBuilderWithHeadersTest {
adapterStore.close();
}
}
+
+ @Test
+ public void shouldHandleRangeQuery() {
+ // KIP-1356: the native header store now serves RangeQuery
(header-stripped results). A range
+ // query reads the underlying store directly (never the cache), so
caching on/off makes no
+ // difference here; use a store-served (caching-disabled) build,
matching the other range tests.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+
+ final QueryResult<KeyValueIterator<String, String>> result =
+ store.query(RangeQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+ assertTrue(result.isSuccess(), "Expected RangeQuery to be handled
on the native header store");
+ try (KeyValueIterator<String, String> iterator =
result.getResult()) {
+ assertTrue(iterator.hasNext(), "Expected the stored record to
be returned");
+ final KeyValue<String, String> keyValue = iterator.next();
+ assertEquals("k", keyValue.key);
+ assertEquals("v", keyValue.value);
+ assertFalse(iterator.hasNext());
+ }
+ assertNotNull(result.getPosition(), "Expected position to be set");
+ } finally {
+ store.close();
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"NATIVE", "IN_MEMORY"})
+ public void
shouldReturnHeadersForTimestampedRangeWithHeadersQueryOnHeaderPersistingStore(final
StoreType storeType) {
+ // Range counterpart of
shouldReturnHeadersForTimestampedKeyWithHeadersQueryOnHeaderPersistingStore.
+ // A range query reads the underlying store (it never consults the
cache), so use a store-served
+ // (caching-disabled) build; the native and in-memory builds persist
headers, so every element carries them.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, false);
+ try {
+ final Headers headers = headersWith("h", "x");
+ store.put("k", ValueTimestampHeaders.make("v", 123L, headers));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess(), "Expected
TimestampedRangeWithHeadersQuery to succeed");
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ assertTrue(iterator.hasNext());
+ final ReadOnlyRecord<String, String> record = iterator.next();
+ assertEquals("k", record.key());
+ assertEquals("v", record.value());
+ assertEquals(123L, record.timestamp());
+ assertEquals(headers, record.headers());
+ // The IQ result is a read-only snapshot: its headers are
immutable.
+ assertThrows(IllegalStateException.class, () ->
record.headers().add("new", new byte[0]),
+ "IQ result headers should be read-only");
+ assertFalse(iterator.hasNext());
+ }
+ assertNotNull(result.getPosition(), "Expected position to be set");
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldReturnEmptyHeadersForTimestampedRangeWithHeadersQueryOnAdapterStore() {
+ // The timestamped adapter keeps the timestamp but drops headers on
write. A range query reads
+ // the underlying store (never the cache), so the headers always come
back empty (never null) --
+ // unlike the point query, whose warm-cache read can still return them.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.ADAPTER, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ assertTrue(iterator.hasNext());
+ final ReadOnlyRecord<String, String> record = iterator.next();
+ assertEquals("k", record.key());
+ assertEquals("v", record.value());
+ assertEquals(123L, record.timestamp());
+ assertEquals(new RecordHeaders(), record.headers());
+ assertFalse(iterator.hasNext());
+ }
+ assertNotNull(result.getPosition(), "Expected position to be set");
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldThrowForTimestampedRangeWithHeadersQueryOnPlainLegacyStore() {
+ // A plain (non-timestamped) legacy supplier surfaces every entry with
timestamp = -1, which
+ // cannot be represented as a ReadOnlyRecord. A range query reads the
underlying store, so the
+ // failure surfaces while iterating (there is no cache-served success
path like the point query).
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.PLAIN_ADAPTER, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess(), "The range query itself succeeds;
the failure surfaces while iterating");
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ final StreamsException exception =
assertThrows(StreamsException.class, iterator::next,
+ "An entry with ts=-1 cannot be represented as a
ReadOnlyRecord");
+ assertTrue(exception.getMessage().contains("timestamp"),
+ "Expected exception message to mention the timestamp: " +
exception.getMessage());
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldThrowForNegativeStoredTimestampForTimestampedRangeWithHeadersQuery() {
+ // A caller can store a negative timestamp directly. Unlike the point
query (which fails the whole
+ // query with STORE_EXCEPTION), a lazily-evaluated range iterator has
already been returned, so the
+ // failure surfaces by throwing while advancing.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", -1L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ final StreamsException exception =
assertThrows(StreamsException.class, iterator::next);
+ assertTrue(exception.getMessage().contains("timestamp"),
+ "Expected exception message to mention the timestamp: " +
exception.getMessage());
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldCollectExecutionInfoForTimestampedRangeWithHeadersQueryWhenRequested() {
+ // The range query, run through the metered handler with execution
info enabled, must carry both
+ // the wrapped store's entry and the metered handler's entry. (Unlike
the point query there is no
+ // query-level failure path -- a negative timestamp surfaces while
iterating, not as a failed result.)
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(true));
+
+ assertTrue(result.isSuccess());
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ final String info = String.join("\n",
result.getExecutionInfo());
+ assertTrue(
+
info.contains(RocksDBTimestampedStoreWithHeaders.class.getName())
+ &&
info.contains(MeteredTimestampedKeyValueStoreWithHeaders.class.getName()),
+ "execution info missing an entry: " + info);
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldNotCollectExecutionInfoForTimestampedRangeWithHeadersQueryWhenNotRequested()
{
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecordIterator<String, String>> result =
+ store.query(TimestampedRangeWithHeadersQuery.withNoBounds(),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ try (ReadOnlyRecordIterator<String, String> iterator =
result.getResult()) {
+ assertTrue(result.getExecutionInfo().isEmpty(), "Expected no
execution info: " + result.getExecutionInfo());
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void
shouldReturnIdenticalRangeResultsForNativeAndAdapterBuiltStores() {
+ // Build-path parity for the existing (header-stripped) range query
types. Range queries read the
+ // underlying store (never the cache), so use a store-served
(caching-disabled) build. Multiple
+ // keys pin that ordering (not just per-key equality) matches across
both build paths.
+ final TimestampedKeyValueStoreWithHeaders<String, String> nativeStore
= buildAndInitStore(StoreType.NATIVE, false);
+ final TimestampedKeyValueStoreWithHeaders<String, String> adapterStore
= buildAndInitStore(StoreType.ADAPTER, false);
+ try {
+ // Insert in a different order per store: a range result is sorted
by key bytes, not insertion
+ // order, so if either build path were (incorrectly)
insertion-order-dependent, the two stores
+ // would diverge here instead of accidentally matching.
+ for (final String key : List.of("a", "b", "c")) {
+ nativeStore.put(key, ValueTimestampHeaders.make("v-" + key,
123L, headersWith("h", "x")));
+ }
+ for (final String key : List.of("c", "b", "a")) {
+ adapterStore.put(key, ValueTimestampHeaders.make("v-" + key,
123L, headersWith("h", "x")));
+ }
+
+ final List<KeyValue<String, String>> nativeRange =
+ drain(nativeStore.query(RangeQuery.<String,
String>withNoBounds(), PositionBound.unbounded(), new
QueryConfig(false)).getResult());
+ final List<KeyValue<String, String>> adapterRange =
+ drain(adapterStore.query(RangeQuery.<String,
String>withNoBounds(), PositionBound.unbounded(), new
QueryConfig(false)).getResult());
+ assertEquals(3, nativeRange.size(), "Expected all three keys to be
returned");
+ assertEquals(nativeRange, adapterRange,
+ "RangeQuery results (including order) should be identical
across native and adapter build paths");
+
+ final List<KeyValue<String, ValueAndTimestamp<String>>>
nativeTimestampedRange =
+ drain(nativeStore.query(TimestampedRangeQuery.<String,
String>withNoBounds(), PositionBound.unbounded(), new
QueryConfig(false)).getResult());
+ final List<KeyValue<String, ValueAndTimestamp<String>>>
adapterTimestampedRange =
+ drain(adapterStore.query(TimestampedRangeQuery.<String,
String>withNoBounds(), PositionBound.unbounded(), new
QueryConfig(false)).getResult());
+ assertEquals(3, nativeTimestampedRange.size(), "Expected all three
keys to be returned");
+ assertEquals(nativeTimestampedRange, adapterTimestampedRange,
+ "TimestampedRangeQuery results (including order) should be
identical across native and adapter build paths");
+ } finally {
+ nativeStore.close();
+ adapterStore.close();
+ }
+ }
+
+ private static <V> List<KeyValue<String, V>> drain(final
KeyValueIterator<String, V> iterator) {
+ final List<KeyValue<String, V>> out = new ArrayList<>();
+ try (iterator) {
+ while (iterator.hasNext()) {
+ out.add(iterator.next());
+ }
+ }
+ return out;
+ }
}
