aliehsaeedii commented on code in PR #22770:
URL: https://github.com/apache/kafka/pull/22770#discussion_r3554968478
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -683,6 +722,7 @@ public void close() {
final long duration = time.nanoseconds() - startNs;
sensor.record(duration);
iteratorDurationSensor.record(duration);
+ numOpenIterators.decrement();
Review Comment:
The sibling `MeteredTimestampedKeyValueStore` has the same gauge bug this PR
fixes here: its `MeteredTimestampedKeyValueStoreIterator` does
`openIterators.add`/`remove` but never
`numOpenIterators.increment()`/`decrement()`, so `num-open-iterators`
under-counts for plain timestamped stores. Could you please open a jira ticket
for that too? Maybe not related to your KIP, but just as a bug.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -542,6 +546,40 @@ private <R> QueryResult<R> runTimestampedRangeQuery(
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;
Review Comment:
This failure branch (wrapped query returns a failure) has no test — every
range-with-headers test asserts `isSuccess()`. Minor, but a failed-rawResult
case would close it (the sibling handlers share the same untested else).
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java:
##########
@@ -615,4 +605,205 @@ public void
shouldReturnIdenticalResultsForNativeAndAdapterBuiltStores(final boo
adapterStore.close();
}
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldHandleRangeQuery(final boolean cachingEnabled) {
Review Comment:
This asserts only success + an empty result from an empty store, so it can't
catch wrong range data — it's really a 'handled, not UNKNOWN_QUERY_TYPE' smoke
test. Consider putting a record and asserting it comes back.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java:
##########
@@ -188,57 +197,186 @@ public void
shouldServeCacheHitWhenCachingEnabledAndRecordNotYetFlushed() throws
}
@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.
+ startStreamsWithHeadersStore();
- produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
+ // 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));
Review Comment:
The bounded ranges (withRange/withLowerBound/withUpperBound) are only
exercised in the default order. Add one descending bounded range so bounds +
DESCENDING is covered end-to-end (only the full scan tests descending today).
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java:
##########
@@ -615,4 +605,205 @@ public void
shouldReturnIdenticalResultsForNativeAndAdapterBuiltStores(final boo
adapterStore.close();
}
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldHandleRangeQuery(final boolean cachingEnabled) {
+ // KIP-1356: the native header store now serves RangeQuery
(header-stripped results), matching
+ // the adapter build path (caching on or off).
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, cachingEnabled);
+ try {
+ 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()) {
+ assertFalse(iterator.hasNext(), "Expected empty result from an
empty store");
+ }
+ 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()) {
+ assertThrows(StreamsException.class, iterator::next,
Review Comment:
The negative-timestamp `StreamsException` is asserted only by type, never by
message. If the message text matters, assert a fragment of it; otherwise fine
to leave. Minor.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java:
##########
@@ -188,57 +197,186 @@ public void
shouldServeCacheHitWhenCachingEnabledAndRecordNotYetFlushed() throws
}
@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
Review Comment:
No integration test runs a range query with caching ENABLED. Since
`CachingKeyValueStoreWithHeaders.query()` forwards straight to the store
(bypassing the cache), a caching-enabled range query won't see writes not yet
flushed — worth a test that locks in that behavior (the point query has its
cache-hit counterpart).
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java:
##########
@@ -188,57 +197,186 @@ public void
shouldServeCacheHitWhenCachingEnabledAndRecordNotYetFlushed() throws
}
@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.
+ startStreamsWithHeadersStore();
- produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
+ // 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"));
+ }
- final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
+ @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.
+ startStreamsWithPlainSupplierStore();
+
+ 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);
+ }
+ }
+
+ 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 startStreamsWithHeadersStore() 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);
+ startStreamsWithHeadersStore(false);
}
- private void startStreams(final boolean cachingEnabled) throws Exception {
- final StreamsBuilder builder = new StreamsBuilder();
+ private void startStreamsWithHeadersStore(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(),
+ HeadersStoreWriterProcessor::new);
+ }
- kafkaStreams = new KafkaStreams(builder.build(), props());
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+ private void startStreamsWithNonHeadersStore() 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()),
+ PlainStoreWriterProcessor::new);
+ }
+
+ private void startStreamsWithPlainSupplierStore() throws Exception {
Review Comment:
The ADAPTER config isn't covered end-to-end: a WithHeaders builder over a
plain *timestamped* supplier (persistentTimestampedKeyValueStore) keeps
timestamps but drops headers, so a range query returns empty (not null) headers
— a distinct outcome from the three helpers here, tested only at the store
level. Consider a startStreamsWithAdapterStore helper + a range test asserting
empty headers.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -696,6 +736,91 @@ public K peekNextKey() {
}
}
+ /**
+ * 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.
+ */
+ private class
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator
Review Comment:
This is now the third inner iterator in this file repeating the same
scaffolding — fields, `increment()`/`openIterators.add()` in the ctor, and
sensor-record + `decrement()`/`remove()` in close(). Consider extracting a
shared abstract metered-iterator base so each subclass only implements next().
Optional (it matches the current idiom), but the duplication is growing.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -696,6 +736,91 @@ public K peekNextKey() {
}
}
+ /**
+ * 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.
+ */
+ private class
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator
+ implements ReadOnlyRecordIterator<K, V>, MeteredIterator {
+
+ private final KeyValueIterator<Bytes, byte[]> iter;
+ private final Sensor sensor;
+ private final long startNs;
+ private final long startTimestampMs;
+ private final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer;
+
+ private
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator(
+ final KeyValueIterator<Bytes, byte[]> iter,
+ final Sensor sensor,
+ final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer
+ ) {
+ this.iter = iter;
+ this.sensor = sensor;
+ this.valueTimestampHeadersDeserializer =
valueTimestampHeadersDeserializer;
+ this.startNs = time.nanoseconds();
+ this.startTimestampMs = time.milliseconds();
+ numOpenIterators.increment();
+ openIterators.add(this);
+ }
+
+ @Override
+ public long startTimestamp() {
+ return startTimestampMs;
+ }
+
+ @Override
+ 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(
Review Comment:
This throw leaves the iterator open. For a plain/NO_TIMESTAMP store the
throw is an expected outcome of normal iteration, so a caller that catches it
and abandons the iterator leaks the underlying RocksDB iterator and permanently
inflates `num-open-iterators`. Worth documenting that the caller must close in
a finally/try-with-resources.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -639,6 +677,7 @@ private
MeteredTimestampedKeyValueStoreWithHeadersQueryIterator(
this.startNs = time.nanoseconds();
this.startTimestampMs = time.milliseconds();
this.returnPlainValue = returnPlainValue;
+ numOpenIterators.increment();
Review Comment:
This metric fix ships without a regression test. The only
`num-open-iterators` test drives `all()`, which already tracked the gauge on
trunk — nothing asserts the gauge moves for the range/IQ query iterator you fix
here (or the new ReadOnlyRecordIterator). Add a 0→1→0 check over a range query
iterator.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilderWithHeadersTest.java:
##########
@@ -615,4 +605,205 @@ public void
shouldReturnIdenticalResultsForNativeAndAdapterBuiltStores(final boo
adapterStore.close();
}
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void shouldHandleRangeQuery(final boolean cachingEnabled) {
+ // KIP-1356: the native header store now serves RangeQuery
(header-stripped results), matching
+ // the adapter build path (caching on or off).
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, cachingEnabled);
+ try {
+ 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()) {
+ assertFalse(iterator.hasNext(), "Expected empty result from an
empty store");
+ }
+ 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()) {
+ assertThrows(StreamsException.class, iterator::next,
+ "An entry with ts=-1 cannot be represented as a
ReadOnlyRecord");
+ }
+ } 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()) {
+ assertThrows(StreamsException.class, iterator::next);
+ }
+ } 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() {
Review Comment:
Parity-only with a single record: it checks native == adapter but won't
catch a bug symmetric across both paths, and doesn't pin ordering or
multi-element results. Consider a few keys with an expected order.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeadersTest.java:
##########
@@ -302,6 +305,39 @@ public void
shouldGetRangeFromInnerStoreAndRecordRangeMetric() {
assertTrue((Double) metric.metricValue() > 0);
}
+ @Test
+ public void
shouldPropagateBoundsAndDescendingOrderForTimestampedRangeWithHeadersQuery() {
Review Comment:
These two capture tests only cover `lowerBound`+DESCENDING and
`upperBound`+ASCENDING. The default `ResultOrder.ANY` (no order builder) and
the both-bounds `withRange` case aren't asserted here — add a
`withNoBounds`/`withRange` capture check so the ANY path is pinned at this
level.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredTimestampedKeyValueStoreWithHeaders.java:
##########
@@ -696,6 +736,91 @@ public K peekNextKey() {
}
}
+ /**
+ * 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.
+ */
+ private class
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator
+ implements ReadOnlyRecordIterator<K, V>, MeteredIterator {
+
+ private final KeyValueIterator<Bytes, byte[]> iter;
+ private final Sensor sensor;
+ private final long startNs;
+ private final long startTimestampMs;
+ private final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer;
+
+ private
MeteredTimestampedKeyValueStoreWithHeadersReadOnlyRecordIterator(
+ final KeyValueIterator<Bytes, byte[]> iter,
+ final Sensor sensor,
+ final Function<byte[], ValueTimestampHeaders<V>>
valueTimestampHeadersDeserializer
+ ) {
+ this.iter = iter;
+ this.sensor = sensor;
+ this.valueTimestampHeadersDeserializer =
valueTimestampHeadersDeserializer;
+ this.startNs = time.nanoseconds();
+ this.startTimestampMs = time.milliseconds();
+ numOpenIterators.increment();
+ openIterators.add(this);
+ }
+
+ @Override
+ public long startTimestamp() {
+ return startTimestampMs;
+ }
+
+ @Override
+ 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;
+ }
+
+ @Override
+ public void close() {
+ try {
+ iter.close();
+ } finally {
+ final long duration = time.nanoseconds() - startNs;
+ sensor.record(duration);
+ iteratorDurationSensor.record(duration);
+ numOpenIterators.decrement();
Review Comment:
No test covers the gauge on the misuse paths: a second `close()`
double-decrements `numOpenIterators`, and a `next()` that throws and isn't
closed never decrements. If you want the gauge behavior pinned, add a test for
each.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java:
##########
@@ -188,57 +197,186 @@ public void
shouldServeCacheHitWhenCachingEnabledAndRecordNotYetFlushed() throws
}
@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.
+ startStreamsWithHeadersStore();
- produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
+ // 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));
Review Comment:
No range test asserts an empty result (a bound matching no keys). A quick
empty-range case would cover the no-match path.
--
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]
