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 eb7096dcf65 KAFKA-20703: IQv2 TimestampedKeyWithHeadersQuery for
headers-aware key-value stores (KIP-1356) (#22666)
eb7096dcf65 is described below
commit eb7096dcf65b422bf346ed46fb8a9ebfc2612e72
Author: Jess668 <[email protected]>
AuthorDate: Wed Jul 8 04:08:00 2026 -0400
KAFKA-20703: IQv2 TimestampedKeyWithHeadersQuery for headers-aware
key-value stores (KIP-1356) (#22666)
This PR is the **first increment of KIP-1356** — the key/value point
query — for `TimestampedKeyValueStoreWithHeaders`. The
range/window/session header query types described in the KIP will
follow in subsequent PRs.
## Summary
1. **`TimestampedKeyWithHeadersQuery<K, V>`** (new, `@Evolving`,
`org.apache.kafka.streams.query`) —
the headers-aware parallel of `TimestampedKeyQuery`, returning a
`ReadOnlyRecord` that carries the
key, value, timestamp, and headers.
2. **Native header store serves basic IQv2 queries.**
`RocksDBTimestampedStoreWithHeaders` no longer
overrides `query(...)` to return `UNKNOWN_QUERY_TYPE`; it inherits
`RocksDBStore.query()`, so the
existing query types (`KeyQuery`, `RangeQuery`, …) behave identically
on both build paths.
3. **Header store caching participates in IQv2 point queries.**
`CachingKeyValueStoreWithHeaders` no
longer overrides `query(...)` to bypass the cache; it inherits
`CachingKeyValueStore.query()`, so
point queries consult the record cache (read-your-writes) and honor
`skipCache`.
4. **`skipCache` is honored for the new query.**
`MeteredTimestampedKeyValueStoreWithHeaders` forwards `isSkipCache()`
from `TimestampedKeyWithHeadersQuery` onto the raw `KeyQuery` it builds,
so the caching layer can honor it. (Making `skipCache()` work for the
existing `KeyQuery`/`TimestampedKeyQuery` on the metered stores is a
pre-existing, cross-cutting no-op fix — out of scope here, to be
addressed in a follow-up
[JIRA](https://issues.apache.org/jira/browse/KAFKA-20776).)
The metered header store services `TimestampedKeyWithHeadersQuery` by
forwarding a raw byte-level `KeyQuery` to the wrapped store and
deserializing the stored `ValueTimestampHeaders` into a `Record` (a
`ReadOnlyRecord`); an absent or tombstoned key yields a `null` result.
## Testing
- `MeteredTimestampedKeyValueStoreWithHeadersTest` — `skipCache` is
propagated to the forwarded
`KeyQuery` for `TimestampedKeyWithHeadersQuery`.
- `RocksDBTimestampedStoreWithHeadersTest` — the native store serves
`KeyQuery` (previously `UNKNOWN_QUERY_TYPE`); a genuinely unsupported
query and `RangeQuery` still returns `UNKNOWN_QUERY_TYPE`.
- `TimestampedKeyValueStoreBuilderWithHeadersTest` — IQv2 query handling
parameterized over build path
(native / adapter / in-memory) × caching on/off, plus a
native-vs-adapter result-parity test.
- `IQv2HeadersStoreIntegrationTest` — end-to-end value/timestamp/headers
round-trip,
empty-headers, tombstone/absent, `UNKNOWN_QUERY_TYPE` against a
non-headers store, and a
caching-enabled cache-hit (read-your-writes) case.
Reviewers: Alieh Saeedi <[email protected]>, TengYao Chi
<[email protected]>
---
.../IQv2HeadersStoreIntegrationTest.java | 347 +++++++++
.../query/TimestampedKeyWithHeadersQuery.java | 100 +++
.../internals/CachingKeyValueStoreWithHeaders.java | 20 +-
...MeteredTimestampedKeyValueStoreWithHeaders.java | 78 +-
.../RocksDBTimestampedStoreWithHeaders.java | 11 +-
...redTimestampedKeyValueStoreWithHeadersTest.java | 34 +
.../RocksDBTimestampedStoreWithHeadersTest.java | 53 +-
...stampedKeyValueStoreBuilderWithHeadersTest.java | 808 ++++++++-------------
8 files changed, 918 insertions(+), 533 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
new file mode 100644
index 00000000000..f8eb62da2e1
--- /dev/null
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2HeadersStoreIntegrationTest.java
@@ -0,0 +1,347 @@
+/*
+ * 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.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.serialization.StringSerializer;
+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.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.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.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.state.StoreBuilder;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+import org.apache.kafka.streams.state.ValueTimestampHeaders;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+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.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.
+ */
+@Tag("integration")
+public class IQv2HeadersStoreIntegrationTest {
+
+ private static final String STORE_NAME = "headers-store";
+
+ private static final EmbeddedKafkaCluster CLUSTER = new
EmbeddedKafkaCluster(1);
+
+ private static final Headers HEADERS = new RecordHeaders()
+ .add("source", "test".getBytes())
+ .add("version", "1.0".getBytes());
+
+ private String inputStream;
+ private String outputStream;
+ private long baseTimestamp;
+ private long commitIntervalMs = 1000L;
+ private Position inputPosition;
+ private KafkaStreams kafkaStreams;
+ private TestInfo testInfo;
+
+ @BeforeAll
+ public static void before() throws IOException {
+ CLUSTER.start();
+ }
+
+ @AfterAll
+ public static void after() {
+ CLUSTER.stop();
+ }
+
+ @BeforeEach
+ public void beforeTest(final TestInfo testInfo) throws
InterruptedException {
+ this.testInfo = testInfo;
+ final String uniqueTestName = safeUniqueTestName(testInfo);
+ inputStream = "input-stream-" + uniqueTestName;
+ outputStream = "output-stream-" + uniqueTestName;
+ CLUSTER.createTopic(inputStream);
+ CLUSTER.createTopic(outputStream);
+ baseTimestamp = CLUSTER.time.milliseconds();
+ inputPosition = Position.emptyPosition();
+ }
+
+ @AfterEach
+ public void afterTest() {
+ if (kafkaStreams != null) {
+ kafkaStreams.close(Duration.ofSeconds(30L));
+ kafkaStreams.cleanUp();
+ }
+ }
+
+ @Test
+ public void shouldHandleTimestampedKeyWithHeadersQuery() throws Exception {
+ startStreams();
+
+ // key 1 has headers, key 2 has empty headers, key 3 is tombstoned
(null value)
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
+ KeyValue.pair(1, "a0"));
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp + 1, new
RecordHeaders(),
+ KeyValue.pair(2, "b0"));
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp + 2, HEADERS,
+ KeyValue.pair(3, "c0"), KeyValue.pair(3, null));
+
+ // key 1: key + value + timestamp + headers round-trip
+ final ReadOnlyRecord<Integer, String> result1 = query(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);
+ 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));
+
+ // never-written key -> null result
+ assertNull(query(999));
+ }
+
+ @Test
+ public void shouldServeCacheHitWhenCachingEnabledAndRecordNotYetFlushed()
throws Exception {
+ // Use a very large commit interval so the processed record is never
committed/flushed during
+ // the test: it lives only in the record cache (the persistent RocksDB
layer stays empty). A
+ // 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);
+
+ 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);
+ // 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));
+ assertEquals(Integer.valueOf(1), result.key());
+ assertEquals("a0", result.value());
+ assertEquals(baseTimestamp, result.timestamp());
+ assertEquals(HEADERS, result.headers());
+ }
+
+ @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()));
+
+ kafkaStreams = new KafkaStreams(builder.build(), props());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+
+ produceDataToTopicWithHeaders(inputStream, baseTimestamp, HEADERS,
KeyValue.pair(1, "a0"));
+
+ final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
+ inStore(STORE_NAME)
+ .withQuery(TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(1))
+ .withPositionBound(PositionBound.at(inputPosition));
+ final StateQueryResult<ReadOnlyRecord<Integer, String>> result =
+ IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
+
+ assertTrue(result.getOnlyPartitionResult().isFailure());
+ assertEquals(FailureReason.UNKNOWN_QUERY_TYPE,
result.getOnlyPartitionResult().getFailureReason());
+ }
+
+ private void startStreams() 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);
+ }
+
+ private void startStreams(final boolean cachingEnabled) throws Exception {
+ final StreamsBuilder builder = new StreamsBuilder();
+ 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()));
+
+ kafkaStreams = new KafkaStreams(builder.build(), props());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
+ }
+
+ private ReadOnlyRecord<Integer, String> query(final int key) {
+ final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
+ inStore(STORE_NAME)
+ .withQuery(TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(key))
+ .withPositionBound(PositionBound.at(inputPosition));
+ // Retry until the store has caught up to the produced input position;
freshness comes from the
+ // IQv2 position mechanism rather than from output-topic consumption.
+ final StateQueryResult<ReadOnlyRecord<Integer, String>> result =
+ IntegrationTestUtils.iqv2WaitForResult(kafkaStreams, request);
+ // getOnlyPartitionResult() returns null when the single partition
result is a successful
+ // null (tombstoned / absent key), which we surface to the caller as a
null lookup.
+ final QueryResult<ReadOnlyRecord<Integer, String>> onlyResult =
result.getOnlyPartitionResult();
+ return onlyResult == null ? null : onlyResult.getResult();
+ }
+
+ private ReadOnlyRecord<Integer, String> querySkipCache(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.
+ final StateQueryRequest<ReadOnlyRecord<Integer, String>> request =
+ inStore(STORE_NAME).withQuery(
+ TimestampedKeyWithHeadersQuery.<Integer,
String>withKey(key).skipCache());
+ final StateQueryResult<ReadOnlyRecord<Integer, String>> result =
kafkaStreams.query(request);
+ final QueryResult<ReadOnlyRecord<Integer, String>> onlyResult =
result.getOnlyPartitionResult();
+ return onlyResult == null ? null : onlyResult.getResult();
+ }
+
+ private Properties props() {
+ final String safeTestName = safeUniqueTestName(testInfo);
+ final Properties streamsConfiguration = new Properties();
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "app-" +
safeTestName);
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
+ streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
commitIntervalMs);
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
+ return streamsConfiguration;
+ }
+
+ @SuppressWarnings("varargs")
+ @SafeVarargs
+ private final void produceDataToTopicWithHeaders(final String topic,
+ final long timestamp,
+ final Headers headers,
+ final KeyValue<Integer,
String>... keyValues) {
+ final Properties producerConfig =
+ TestUtils.producerConfig(CLUSTER.bootstrapServers(),
IntegerSerializer.class, StringSerializer.class);
+ final List<Future<RecordMetadata>> futures = new LinkedList<>();
+ try (final Producer<Integer, String> producer = new
KafkaProducer<>(producerConfig)) {
+ for (final KeyValue<Integer, String> keyValue : keyValues) {
+ futures.add(producer.send(
+ new ProducerRecord<>(topic, null, timestamp, keyValue.key,
keyValue.value, headers)));
+ }
+ producer.flush();
+ for (final Future<RecordMetadata> future : futures) {
+ try {
+ final RecordMetadata metadata = future.get(60,
TimeUnit.SECONDS);
+ // Track the produced input Position so queries can bound
on it (IQv2 freshness).
+ inputPosition = inputPosition.withComponent(
+ metadata.topic(), metadata.partition(),
metadata.offset());
+ } catch (final Exception e) {
+ throw new RuntimeException("Failed to produce test record
to " + topic, e);
+ }
+ }
+ }
+ }
+
+ private static class HeadersStoreWriterProcessor implements
Processor<Integer, String, Integer, String> {
+ private ProcessorContext<Integer, String> context;
+ private TimestampedKeyValueStoreWithHeaders<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Integer, String> context) {
+ this.context = context;
+ store = context.getStateStore(STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ 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 ProcessorContext<Integer, String> context;
+ private TimestampedKeyValueStore<Integer, String> store;
+
+ @Override
+ public void init(final ProcessorContext<Integer, String> context) {
+ this.context = context;
+ store = context.getStateStore(STORE_NAME);
+ }
+
+ @Override
+ public void process(final Record<Integer, String> record) {
+ store.put(
+ record.key(),
+ ValueAndTimestamp.make(record.value(), record.timestamp()));
+ context.forward(record);
+ }
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/query/TimestampedKeyWithHeadersQuery.java
b/streams/src/main/java/org/apache/kafka/streams/query/TimestampedKeyWithHeadersQuery.java
new file mode 100644
index 00000000000..cbcca190385
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/query/TimestampedKeyWithHeadersQuery.java
@@ -0,0 +1,100 @@
+/*
+ * 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.TimestampedKeyValueStoreWithHeaders;
+
+import java.util.Objects;
+
+/**
+ * Interactive query for retrieving a single record, including its record
headers, based on its key
+ * from a {@link TimestampedKeyValueStoreWithHeaders}.
+ *
+ * <p>This is the headers-aware parallel of {@link TimestampedKeyQuery}: it
returns a
+ * {@link ReadOnlyRecord} carrying the key, value, timestamp, and headers,
whereas
+ * {@link TimestampedKeyQuery} returns a {@link
org.apache.kafka.streams.state.ValueAndTimestamp}
+ * (value and timestamp only, no key or headers).
+ *
+ * <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 result
depends on who serves
+ * the read: with caching enabled, a not-yet-evicted entry is served from the
record cache and
+ * returns the written value, timestamp, and headers (read-your-writes); once
served from the
+ * underlying store, the headers are gone. Over a timestamped legacy supplier
(e.g.
+ * {@code Stores.persistentTimestampedKeyValueStore}) the store-served read
succeeds with empty
+ * {@code headers()}; over a plain legacy supplier (e.g. {@code
Stores.persistentKeyValueStore}),
+ * which keeps no timestamp either, the store-served read has no representable
timestamp and the
+ * query fails with {@link FailureReason#STORE_EXCEPTION}.
+ *
+ * <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}.
+ *
+ * @param <K> Type of keys
+ * @param <V> Type of values
+ */
+@Evolving
[email protected]
+public final class TimestampedKeyWithHeadersQuery<K, V> implements
Query<ReadOnlyRecord<K, V>> {
+
+ private final K key;
+ private final boolean skipCache;
+
+ private TimestampedKeyWithHeadersQuery(final K key, final boolean
skipCache) {
+ this.key = key;
+ this.skipCache = skipCache;
+ }
+
+ /**
+ * Creates a query that will retrieve the record (value, timestamp, and
headers) identified by
+ * {@code key} if it exists (or {@code null} otherwise).
+ * @param key The key to retrieve
+ * @param <K> The type of the key
+ * @param <V> The type of the value that will be retrieved
+ */
+ public static <K, V> TimestampedKeyWithHeadersQuery<K, V> withKey(final K
key) {
+ Objects.requireNonNull(key, "the key should not be null");
+ return new TimestampedKeyWithHeadersQuery<>(key, false);
+ }
+
+ /**
+ * Specifies that the cache should be skipped during query evaluation.
This means, that the query will always
+ * get forwarded to the underlying store.
+ */
+ public TimestampedKeyWithHeadersQuery<K, V> skipCache() {
+ return new TimestampedKeyWithHeadersQuery<>(key, true);
+ }
+
+ /**
+ * Return the key that was specified for this query.
+ *
+ * @return The key that was specified for this query.
+ */
+ public K key() {
+ return key;
+ }
+
+ /**
+ * The flag whether to skip the cache or not during query evaluation.
+ */
+ public boolean isSkipCache() {
+ return skipCache;
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreWithHeaders.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreWithHeaders.java
index 47c9c10c45c..f2609803ed7 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreWithHeaders.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreWithHeaders.java
@@ -18,26 +18,20 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.utils.Bytes;
-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.KeyValueStore;
/**
- * A caching key-value store with headers is a caching key-value store that
only forwards the query to the
- * wrapped store.
+ * A caching key-value store with headers.
+ *
+ * <p>It inherits {@link CachingKeyValueStore}'s IQv2 {@code query(...)}
handling: point queries
+ * ({@code KeyQuery}) consult the record cache (honoring {@code skipCache})
and fall back to the
+ * wrapped store on a miss, while other query types are forwarded to the
wrapped store. The cached
+ * value bytes are the serialized {@code ValueTimestampHeaders}; the metered
layer performs the
+ * header-aware deserialization, so no override is needed here.
*/
public class CachingKeyValueStoreWithHeaders extends CachingKeyValueStore {
CachingKeyValueStoreWithHeaders(final KeyValueStore<Bytes, byte[]>
underlying) {
super(underlying, CacheType.TIMESTAMPED_KEY_VALUE_STORE_WITH_HEADERS);
}
-
- @Override
- public <R> QueryResult<R> query(final Query<R> query,
- final PositionBound positionBound,
- final QueryConfig config) {
- return wrapped().query(query, positionBound, config);
- }
}
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 8d10c6ccd25..98b6a024b22 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,8 +26,11 @@ 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.processor.api.ReadOnlyRecord;
+import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.SerdeGetter;
+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;
@@ -36,6 +39,7 @@ 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.TimestampedKeyQuery;
+import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
import org.apache.kafka.streams.query.TimestampedRangeQuery;
import org.apache.kafka.streams.query.internals.InternalQueryResultUtil;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -91,6 +95,10 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K, V>
TimestampedKeyQuery.class,
(query, positionBound, config, store) ->
runTimestampedKeyQuery(query, positionBound, config)
),
+ mkEntry(
+ TimestampedKeyWithHeadersQuery.class,
+ (query, positionBound, config, store) ->
runTimestampedKeyWithHeadersQuery(query, positionBound, config)
+ ),
mkEntry(
RangeQuery.class,
(query, positionBound, config, store) -> runRangeQuery(query,
positionBound, config)
@@ -286,10 +294,6 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
/**
* Executes a query against this store.
*
- * <p>Note: Query results do NOT include headers, even though headers are
- * preserved in the underlying store. This behavior provides compatibility
- * with existing IQv2 APIs that operate on timestamped stores.
- *
* @param query the query to execute
* @param positionBound the position bound
* @param config the query configuration
@@ -381,6 +385,72 @@ public class MeteredTimestampedKeyValueStoreWithHeaders<K,
V>
return result;
}
+ @SuppressWarnings("unchecked")
+ private <R> QueryResult<R> runTimestampedKeyWithHeadersQuery(
+ final Query<R> query,
+ final PositionBound positionBound,
+ final QueryConfig config
+ ) {
+ final QueryResult<R> result;
+ final TimestampedKeyWithHeadersQuery<K, V> typedKeyQuery =
(TimestampedKeyWithHeadersQuery<K, V>) query;
+
+ // Forward a raw byte-level KeyQuery to the wrapped store, propagating
skipCache so the caching
+ // layer can honor it; the result bytes are the serialized
ValueTimestampHeaders, which we
+ // deserialize below to recover value, timestamp, and headers.
+ // The existing KeyQuery/TimestampedKeyQuery handlers do not yet
propagate skipCache across the
+ // metered stores; that general fix is tracked in KAFKA-20776.
+ KeyQuery<Bytes, byte[]> rawKeyQuery =
KeyQuery.withKey(serializeKey(typedKeyQuery.key(), internalContext.headers()));
+ if (typedKeyQuery.isSkipCache()) {
+ rawKeyQuery = rawKeyQuery.skipCache();
+ }
+ final QueryResult<byte[]> rawResult = wrapped().query(rawKeyQuery,
positionBound, config);
+ if (rawResult.isSuccess()) {
+ final Function<byte[], ValueTimestampHeaders<V>> deserializer =
StoreQueryUtils.deserializeValue(serdes, wrapped());
+ final ValueTimestampHeaders<V> valueTimestampHeaders =
deserializer.apply(rawResult.getResult());
+ if (valueTimestampHeaders != null &&
valueTimestampHeaders.timestamp() < 0) {
+ // The result is modeled as a Record, whose constructor
rejects negative timestamps. A
+ // negative stored timestamp cannot arise from the normal
record-driven flow (the PAPI
+ // Record a processor stores already forbids it), so it
indicates corrupted/unexpected
+ // store state; surface it as a failed result rather than
letting `new Record<>` throw
+ // out of query().
+ final QueryResult<ReadOnlyRecord<K, V>> failure =
QueryResult.forFailure(
+ FailureReason.STORE_EXCEPTION,
+ "Stored record for the queried key has a negative
timestamp ("
+ + valueTimestampHeaders.timestamp() + "); cannot
construct a ReadOnlyRecord.");
+ // Preserve the wrapped store's execution info (empty unless
collectExecutionInfo is set),
+ // matching the success path and the raw-failure path below.
+
rawResult.getExecutionInfo().forEach(failure::addExecutionInfo);
+ failure.setPosition(rawResult.getPosition());
+ result = (QueryResult<R>) failure;
+ } else {
+ // Surface the result as a ReadOnlyRecord (implemented by
Record), keeping the headers.
+ // A null wrapper means the key is absent or tombstoned, which
we surface as a null result.
+ final ReadOnlyRecord<K, V> record;
+ if (valueTimestampHeaders == null) {
+ record = null;
+ } else {
+ final Record<K, V> headerRecord = new Record<>(
+ typedKeyQuery.key(),
+ valueTimestampHeaders.value(),
+ valueTimestampHeaders.timestamp(),
+ valueTimestampHeaders.headers());
+ // An IQ result is a read-only snapshot, so its headers
should be immutable too.
+ // Record copies the headers into a RecordHeaders; mark it
read-only so a caller
+ // cannot mutate the returned Headers.
+ ((RecordHeaders) headerRecord.headers()).setReadOnly();
+ record = headerRecord;
+ }
+ final QueryResult<ReadOnlyRecord<K, V>> typedQueryResult =
+
InternalQueryResultUtil.copyAndSubstituteDeserializedResult(rawResult, record);
+ result = (QueryResult<R>) typedQueryResult;
+ }
+ } else {
+ // the generic type doesn't matter, since failed queries have no
result set.
+ result = (QueryResult<R>) rawResult;
+ }
+ return result;
+ }
+
@SuppressWarnings("unchecked")
private <R> QueryResult<R> runRangeQuery(
final Query<R> query,
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 8f96b182916..d0ad49acfcd 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,6 +18,7 @@
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;
@@ -199,6 +200,14 @@ public class RocksDBTimestampedStoreWithHeaders extends
RocksDBStore implements
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;
@@ -215,4 +224,4 @@ public class RocksDBTimestampedStoreWithHeaders extends
RocksDBStore implements
return result;
}
-}
\ No newline at end of file
+}
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 f6355e7bd6a..568a11a7226 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,6 +36,12 @@ 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.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.TimestampedKeyWithHeadersQuery;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
@@ -43,6 +49,7 @@ import org.apache.kafka.test.KeyValueIteratorStub;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
@@ -143,6 +150,33 @@ public class
MeteredTimestampedKeyValueStoreWithHeadersTest {
metered.init(context, metered);
}
+ // --- skipCache propagation: the TimestampedKeyWithHeadersQuery handler
must forward isSkipCache()
+ // onto the raw KeyQuery it builds, otherwise the caching layer never
sees it. ---
+
+ @Test
+ public void shouldPropagateSkipCacheForTimestampedKeyWithHeadersQuery() {
+ setUp();
+ init();
+
assertTrue(forwardedRawKeyQuery(TimestampedKeyWithHeadersQuery.withKey(KEY).skipCache()).isSkipCache());
+ }
+
+ @Test
+ public void shouldNotSkipCacheForTimestampedKeyWithHeadersQueryByDefault()
{
+ setUp();
+ init();
+
assertFalse(forwardedRawKeyQuery(TimestampedKeyWithHeadersQuery.withKey(KEY)).isSkipCache());
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private KeyQuery<?, ?> forwardedRawKeyQuery(final Query<?> query) {
+ when(inner.query(any(), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn((QueryResult) QueryResult.forResult(null));
+ metered.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+ final ArgumentCaptor<KeyQuery> captor =
ArgumentCaptor.forClass(KeyQuery.class);
+ verify(inner).query(captor.capture(), any(PositionBound.class),
any(QueryConfig.class));
+ return captor.getValue();
+ }
+
@Test
public void shouldDelegateInit() {
setUp();
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 d098696ded7..be048ece610 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
@@ -25,8 +25,10 @@ import
org.apache.kafka.streams.errors.ProcessorStateException;
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.state.KeyValueIterator;
import org.junit.jupiter.api.Test;
@@ -981,17 +983,17 @@ public class RocksDBTimestampedStoreWithHeadersTest
extends RocksDBStoreTest {
}
@Test
- public void shouldReturnUnknownQueryTypeForQuery() {
+ public void shouldReturnUnknownQueryTypeForUnsupportedQuery() {
// Initialize the store
rocksDBStore.init(context, rocksDBStore);
- // Create a query
- final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(new
Bytes("test-key".getBytes()));
+ // A query type the store has no handler for still yields
UNKNOWN_QUERY_TYPE.
+ final Query<Void> unsupportedQuery = new Query<>() { };
final PositionBound positionBound = PositionBound.unbounded();
final QueryConfig config = new QueryConfig(false);
// Execute query
- final QueryResult<byte[]> result = rocksDBStore.query(query,
positionBound, config);
+ final QueryResult<Void> result = rocksDBStore.query(unsupportedQuery,
positionBound, config);
// Verify result indicates unknown query type
assertFalse(result.isSuccess(), "Expected query to fail with unknown
query type");
@@ -1005,6 +1007,49 @@ public class RocksDBTimestampedStoreWithHeadersTest
extends RocksDBStoreTest {
assertNotNull(result.getPosition(), "Expected position to be set");
}
+ @Test
+ public void shouldHandleKeyQuery() {
+ // Initialize the store
+ rocksDBStore.init(context, rocksDBStore);
+
+ // 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: removing the query() override lets the native store serve
KeyQuery (previously
+ // UNKNOWN_QUERY_TYPE), matching the adapter build path. The raw
stored bytes are returned;
+ // the metered store performs the header-aware deserialization.
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+ final QueryResult<byte[]> result =
+ rocksDBStore.query(query, PositionBound.unbounded(), new
QueryConfig(false));
+
+ assertTrue(result.isSuccess(), "Expected KeyQuery to succeed");
+ assertArrayEquals(storedBytes, result.getResult(), "Expected the raw
stored bytes to be returned");
+ assertNotNull(result.getPosition(), "Expected position to be set");
+ }
+
+ @Test
+ public void shouldReturnUnknownQueryTypeForRangeQuery() {
+ // 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.
+ 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"
+ );
+ assertNotNull(result.getPosition(), "Expected position to be set");
+ }
+
@Test
public void shouldCollectExecutionInfoWhenRequested() {
// Initialize the store
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 3c150da4f3d..cccc1174e53 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
@@ -19,11 +19,15 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
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.streams.StreamsConfig;
+import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.api.ReadOnlyRecord;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.query.FailureReason;
import org.apache.kafka.streams.query.KeyQuery;
import org.apache.kafka.streams.query.Position;
@@ -32,29 +36,27 @@ 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.TimestampedKeyQuery;
-import org.apache.kafka.streams.state.HeadersBytesStore;
+import org.apache.kafka.streams.query.TimestampedKeyWithHeadersQuery;
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.TimestampedKeyValueStoreWithHeaders;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.ValueTimestampHeaders;
import org.apache.kafka.test.InternalMockProcessorContext;
-import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
-import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
-import java.util.Properties;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -62,9 +64,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
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.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.lenient;
-import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -228,356 +228,271 @@ public class
TimestampedKeyValueStoreBuilderWithHeadersTest {
assertInstanceOf(PlainToHeadersStoreAdapter.class,
((WrappedStateStore) store).wrapped());
}
- @Test
- public void shouldHandleKeyQueryOnInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
+ //
----------------------------------------------------------------------------------------------
+ // IQv2 query handling for the built header store.
+ //
+ // The header store can be built three ways; each is exercised with
caching on and off through a
+ // single helper that builds the real store chain (Metered -> [Caching] ->
inner) with a real
+ // record cache, then writes and reads real data at the typed (metered)
level:
+ // NATIVE -> RocksDBTimestampedStoreWithHeaders
(persists headers)
+ // ADAPTER -> RocksDBTimestampedStore via
TimestampedToHeadersStoreAdapter (drops headers)
+ // IN_MEMORY -> InMemoryKeyValueStore via the in-memory headers marker
+ //
----------------------------------------------------------------------------------------------
+
+ private enum StoreType { NATIVE, ADAPTER, PLAIN_ADAPTER, IN_MEMORY }
+
+ private KeyValueStore<Bytes, byte[]> innerStore(final StoreType storeType)
{
+ switch (storeType) {
+ case NATIVE: return new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope");
+ case ADAPTER: return new RocksDBTimestampedStore("test-store",
"metrics-scope");
+ case PLAIN_ADAPTER: return new RocksDBStore("test-store",
"metrics-scope");
+ case IN_MEMORY: return new InMemoryKeyValueStore("test-store");
+ default: throw new IllegalArgumentException("unknown store
type: " + storeType);
+ }
+ }
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
+ private TimestampedKeyValueStoreWithHeaders<String, String>
buildAndInitStore(
+ final StoreType storeType,
+ final boolean cachingEnabled) {
+ lenient().when(supplier.name()).thenReturn("test-store");
+ lenient().when(supplier.metricsScope()).thenReturn("metricScope");
+ lenient().when(supplier.get()).thenReturn(innerStore(storeType));
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
+ builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
+ supplier, Serdes.String(), Serdes.String(), new MockTime());
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
+ (cachingEnabled
+ ? builder.withLoggingDisabled().withCachingEnabled()
+ : builder.withLoggingDisabled().withCachingDisabled())
+ .build();
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
+ final ThreadCache cache = new ThreadCache(
+ new LogContext("test "),
+ cachingEnabled ? 10 * 1024 * 1024 : 0,
+ new MockStreamsMetrics(new Metrics()));
final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
+ TestUtils.tempDirectory(), Serdes.String(), Serdes.String(), null,
cache);
+ context.setRecordContext(new ProcessorRecordContext(0L, 0L, 0,
"topic", new RecordHeaders()));
store.init(context, store);
+ return store;
+ }
- try {
- // Put data into the store
- final Headers headers = new RecordHeaders();
- headers.add("key1", "value1".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("test-value",
12345L, headers));
+ private static Headers headersWith(final String key, final String value) {
+ return new RecordHeaders().add(key,
value.getBytes(StandardCharsets.UTF_8));
+ }
- // Verify wrapper type for InMemoryKeyValueStore
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped,
- "Expected wrapper to implement HeadersBytesStore for
InMemoryKeyValueStore");
+ @ParameterizedTest
+ @CsvSource({"NATIVE, true", "NATIVE, false", "ADAPTER, true", "ADAPTER,
false", "IN_MEMORY, true", "IN_MEMORY, false"})
+ public void shouldHandleKeyQuery(final StoreType storeType, final boolean
cachingEnabled) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, cachingEnabled);
+ try {
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
- // Query at typed level - KeyQuery should return just the value
- final KeyQuery<String, String> query =
KeyQuery.withKey("test-key");
- final QueryResult<String> result = store.query(query,
PositionBound.unbounded(), new QueryConfig(false));
+ final QueryResult<String> result =
+ store.query(KeyQuery.withKey("k"), PositionBound.unbounded(),
new QueryConfig(false));
- // Verify IQv2 query result
- assertTrue(result.isSuccess(), "Expected query to succeed on
InMemoryKeyValueStore");
+ assertTrue(result.isSuccess(), "Expected KeyQuery to succeed");
+ assertEquals("v", result.getResult(), "KeyQuery returns the value
only");
assertNotNull(result.getPosition(), "Expected position to be set");
- assertInstanceOf(String.class, result.getResult());
- assertEquals("test-value", result.getResult(), "KeyQuery should
return just the value");
} finally {
store.close();
}
}
- @Test
- public void shouldHandleTimestampedKeyQueryOnInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @CsvSource({"NATIVE, true", "NATIVE, false", "ADAPTER, true", "ADAPTER,
false", "IN_MEMORY, true", "IN_MEMORY, false"})
+ public void shouldHandleTimestampedKeyQuery(final StoreType storeType,
final boolean cachingEnabled) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, cachingEnabled);
try {
- // Put data into the store
- final Headers headers = new RecordHeaders();
- headers.add("key1", "value1".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("test-value",
12345L, headers));
-
- // Verify wrapper type for InMemoryKeyValueStore
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped,
- "Expected wrapper to implement HeadersBytesStore for
InMemoryKeyValueStore");
+ store.put("k", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
- // Query at typed level - TimestampedKeyQuery should return value
+ timestamp
- final TimestampedKeyQuery<String, String> query =
TimestampedKeyQuery.withKey("test-key");
- final QueryResult<ValueAndTimestamp<String>> result =
store.query(query, PositionBound.unbounded(), new QueryConfig(false));
+ final QueryResult<ValueAndTimestamp<String>> result =
+ store.query(TimestampedKeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
- // Verify IQv2 query result
- assertTrue(result.isSuccess(), "Expected query to succeed on
InMemoryKeyValueStore");
+ assertTrue(result.isSuccess(), "Expected TimestampedKeyQuery to
succeed");
+ assertEquals("v", result.getResult().value());
+ assertEquals(123L, result.getResult().timestamp());
assertNotNull(result.getPosition(), "Expected position to be set");
- assertNotNull(result.getResult(), "Expected non-null result");
- assertInstanceOf(ValueAndTimestamp.class, result.getResult());
- assertEquals("test-value", result.getResult().value(),
"TimestampedKeyQuery should return the value");
- assertEquals(12345L, result.getResult().timestamp(),
"TimestampedKeyQuery should return the timestamp");
} finally {
store.close();
}
}
- @Test
- public void shouldReturnPositionFromHeadersStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @CsvSource({"NATIVE, true", "NATIVE, false", "IN_MEMORY, true",
"IN_MEMORY, false"})
+ public void
shouldReturnHeadersForTimestampedKeyWithHeadersQueryOnHeaderPersistingStore(
+ final StoreType storeType, final boolean cachingEnabled) {
+ // The native and in-memory builds both persist headers: the native
store keeps them in its
+ // headers column family, and the in-memory marker stores the value
bytes verbatim (the metered
+ // layer serializes the headers into those bytes). The adapter build
is the one that drops them.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, cachingEnabled);
try {
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final Position position = wrapped.getPosition();
-
- // Verify: Position is returned (should be non-null)
- assertNotNull(position, "Expected non-null position");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
+ final Headers headers = headersWith("h", "x");
+ store.put("k", ValueTimestampHeaders.make("v", 123L, headers));
+
+ final QueryResult<ReadOnlyRecord<String, String>> result =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess(), "Expected
TimestampedKeyWithHeadersQuery to succeed");
+ final ReadOnlyRecord<String, String> record = result.getResult();
+ assertEquals("k", record.key());
+ assertEquals("v", record.value());
+ assertEquals(123L, record.timestamp());
+ // Headers must round-trip on both the persistent path and the
cache path.
+ 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");
+ assertNotNull(result.getPosition(), "Expected position to be set");
} finally {
store.close();
}
}
- @Test
- public void shouldReturnPositionFromAdaptedTimestampedStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStore("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReflectFlushTimingForTimestampedKeyWithHeadersQueryOnAdapterStore(final
boolean cachingEnabled) {
+ // Feeding a non-header supplier into the WithHeaders builder yields
an adapter-built store
+ // (TimestampedToHeadersStoreAdapter over a plain timestamped store),
which cannot persist
+ // headers. The metered layer still serializes headers into the value
bytes, and the record
+ // cache sits above the adapter, so the header result depends on who
serves the read:
+ // - store-served (caching disabled, or -- when caching is enabled
-- once the entry has been
+ // evicted/flushed out of the cache): the read goes through the
adapter, which stripped the
+ // headers on write, so they come back empty (never null) even
though they were written;
+ // - cache-served (caching enabled, entry still warm): returned with
headers intact
+ // (read-your-writes), since the cache holds the full serialized
bytes.
+ // The cache is the only thing preserving headers on this build. The
caching-disabled run reads
+ // through the same adapter a post-eviction cache miss would, so it
covers the store-served
+ // (empty) outcome; the caching-enabled run covers the cache-served
(headers) outcome and then
+ // pins the flush-dependent flip on the same instance: after flushing
the cache, a skipCache read
+ // (which goes to the store, bypassing the cache) returns empty
headers.
+ // value/timestamp round-trip either way.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.ADAPTER, cachingEnabled);
try {
- // Verify adapter is used
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(TimestampedToHeadersStoreAdapter.class, wrapped);
-
- // Get position from adapter (should delegate to underlying store)
- final Position position = wrapped.getPosition();
+ final Headers headers = headersWith("h", "x");
+ store.put("k", ValueTimestampHeaders.make("v", 123L, headers));
+
+ final QueryResult<ReadOnlyRecord<String, String>> result =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess(), "Expected
TimestampedKeyWithHeadersQuery to succeed on an adapter-built store");
+ final ReadOnlyRecord<String, String> record = result.getResult();
+ assertEquals("k", record.key());
+ assertEquals("v", record.value());
+ assertEquals(123L, record.timestamp());
+ // Cache-served (caching enabled, warm) -> written headers;
store-served (caching disabled,
+ // or a post-eviction cache miss) -> empty, since the adapter
stripped them on write.
+ final Headers expectedHeaders = cachingEnabled ? headers : new
RecordHeaders();
+ assertEquals(expectedHeaders, record.headers());
+ assertNotNull(result.getPosition(), "Expected position to be set");
- assertNotNull(position, "Expected non-null position from adapter");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
+ if (cachingEnabled) {
+ // Same instance, flush-dependent flip: flush the record cache
down through the
+ // (header-stripping) adapter to the persistent store, then
read past the cache with
+ // skipCache. The persistent copy has no headers, so the same
key now comes back empty
+ // (value/timestamp still round-trip; the timestamped adapter
keeps the timestamp).
+ ((CachedStateStore<?, ?>) ((WrappedStateStore)
store).wrapped()).flushCache();
+ final QueryResult<ReadOnlyRecord<String, String>> afterFlush =
store.query(
+ TimestampedKeyWithHeadersQuery.<String,
String>withKey("k").skipCache(),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ assertTrue(afterFlush.isSuccess(), "Expected skipCache read to
succeed after flush");
+ assertEquals("v", afterFlush.getResult().value());
+ assertEquals(123L, afterFlush.getResult().timestamp());
+ assertEquals(new RecordHeaders(),
afterFlush.getResult().headers());
+ }
} finally {
store.close();
}
}
- @Test
- public void shouldReturnPositionFromInMemoryStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("in-memory");
- when(supplier.get()).thenReturn(new
InMemoryKeyValueStore("test-store"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReflectFlushTimingForTimestampedKeyWithHeadersQueryOnPlainLegacyStore(final
boolean cachingEnabled) {
+ // A plain (non-timestamped) legacy supplier is wrapped in
PlainToHeadersStoreAdapter, which can
+ // persist neither headers nor a timestamp: on write it keeps only the
raw value, and on read it
+ // rebuilds the header format with empty headers and timestamp = -1.
So the result depends on who
+ // serves the read:
+ // - cache-served (caching enabled, entry still warm): the cache
holds the full serialized bytes
+ // the metered layer wrote, so the query succeeds with the real
value, timestamp, and headers;
+ // - store-served (caching disabled, or once the entry is
evicted/flushed): the value comes back
+ // with timestamp -1, which cannot be represented as a Record, so
the query fails with
+ // STORE_EXCEPTION -- unlike the timestamped adapter (which keeps
the timestamp and just
+ // returns empty headers), a plain store loses the timestamp too.
+ // The caching-enabled run also pins the flip on the same instance:
after flushing the cache, a
+ // skipCache read goes to the store and fails with STORE_EXCEPTION.
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.PLAIN_ADAPTER, cachingEnabled);
try {
- // Verify marker wrapper is used
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(HeadersBytesStore.class, wrapped);
-
- // Get position from marker (should delegate to
InMemoryKeyValueStore)
- final Position position = wrapped.getPosition();
-
- assertNotNull(position, "Expected non-null position from in-memory
store");
- assertTrue(position.isEmpty(), "Expected position to be empty
initially");
+ final Headers headers = headersWith("h", "x");
+ store.put("k", ValueTimestampHeaders.make("v", 123L, headers));
+
+ final QueryResult<ReadOnlyRecord<String, String>> result =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
+
+ if (cachingEnabled) {
+ assertTrue(result.isSuccess(), "Expected a cache-served read
to succeed");
+ final ReadOnlyRecord<String, String> record =
result.getResult();
+ assertEquals("k", record.key());
+ assertEquals("v", record.value());
+ assertEquals(123L, record.timestamp());
+ assertEquals(headers, record.headers());
+
+ // Same instance, flush-dependent flip: flush the cache down
through the plain adapter
+ // (which keeps only the raw value) to the persistent store,
then read past the cache with
+ // skipCache. The store read rebuilds a header record with
timestamp -1, which cannot be a
+ // Record, so the same key now fails with STORE_EXCEPTION.
+ ((CachedStateStore<?, ?>) ((WrappedStateStore)
store).wrapped()).flushCache();
+ final QueryResult<ReadOnlyRecord<String, String>> afterFlush =
store.query(
+ TimestampedKeyWithHeadersQuery.<String,
String>withKey("k").skipCache(),
+ PositionBound.unbounded(),
+ new QueryConfig(false));
+ assertFalse(afterFlush.isSuccess(),
+ "A skipCache read after flush hits the store (ts=-1) and
must fail");
+ assertEquals(FailureReason.STORE_EXCEPTION,
afterFlush.getFailureReason());
+ } else {
+ assertFalse(result.isSuccess(),
+ "A store-served read on a plain build has ts=-1 and must
fail, not return empty headers");
+ assertEquals(FailureReason.STORE_EXCEPTION,
result.getFailureReason());
+ }
+ assertNotNull(result.getPosition(), "Expected position to be set");
} finally {
store.close();
}
}
- @Test
- public void shouldMaintainPositionAcrossOperationsOnHeadersStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldFailTimestampedKeyWithHeadersQueryForNegativeStoredTimestamp(final
boolean cachingEnabled) {
+ // A negative stored timestamp can't arise from the normal
record-driven flow, but a caller can
+ // store one directly. Because the result is modeled as a Record
(whose constructor rejects
+ // negative timestamps), the query must surface a failed result rather
than throw out of query().
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, cachingEnabled);
try {
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
-
- // Get initial position
- final Position initialPosition = wrapped.getPosition();
- assertNotNull(initialPosition, "Expected non-null initial
position");
-
- // Put some data
- store.put("key1", ValueTimestampHeaders.make("value1", 100L, new
RecordHeaders()));
- store.put("key2", ValueTimestampHeaders.make("value2", 200L, new
RecordHeaders()));
+ store.put("k", ValueTimestampHeaders.make("v", -1L,
headersWith("h", "x")));
- // Get position after puts
- final Position afterPutPosition = wrapped.getPosition();
- assertNotNull(afterPutPosition, "Expected non-null position after
puts");
+ final QueryResult<ReadOnlyRecord<String, String>> result =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false));
- // Position object should be the same instance (stores maintain a
single position)
- // The position content might be updated internally by the context
+ assertFalse(result.isSuccess(), "A negative stored timestamp
should yield a failed result, not throw");
+ assertEquals(FailureReason.STORE_EXCEPTION,
result.getFailureReason());
+ assertNotNull(result.getPosition(), "Expected the failure to
preserve the queried partition's position");
} finally {
store.close();
}
}
- private static ThreadCache mockCacheHit() {
- final ThreadCache cache = mock(ThreadCache.class);
- final LRUCacheEntry entry = mock(LRUCacheEntry.class);
- final byte[] entryValue =
"mockEntryValue".getBytes(StandardCharsets.UTF_8);
- lenient().when(entry.value()).thenReturn(entryValue);
- lenient().when(cache.get(any(String.class),
any(Bytes.class))).thenReturn(entry);
- return cache;
- }
-
- private TimestampedKeyValueStoreWithHeaders<String, String>
headersStoreMaybeWithCache(final boolean cachingEnabled) {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- final File dir = TestUtils.tempDirectory();
- final ThreadCache cache = mockCacheHit();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- null,
- cache
- );
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store;
- if (cachingEnabled) {
- store = builder.withLoggingDisabled()
- .withCachingEnabled()
- .build();
- } else {
- store = builder.withLoggingDisabled()
- .withCachingDisabled()
- .build();
- }
-
- store.init(context, store);
- return store;
- }
-
@ParameterizedTest
@ValueSource(booleans = {true, false})
- public void shouldReturnUnknownQueryTypeForKeyQueryOnHeadersStore(final
boolean cachingEnabled) {
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
headersStoreMaybeWithCache(cachingEnabled);
-
+ 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 KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(new
Bytes("test-key".getBytes()));
- final PositionBound positionBound = PositionBound.unbounded();
- final QueryConfig config = new QueryConfig(false);
-
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final QueryResult<byte[]> result = wrapped.query(query,
positionBound, config);
-
- // Verify: Headers store currently returns UNKNOWN_QUERY_TYPE
- assertFalse(result.isSuccess(), "Expected query to fail with
unknown query type");
- assertEquals(
- FailureReason.UNKNOWN_QUERY_TYPE,
- result.getFailureReason(),
- "Expected UNKNOWN_QUERY_TYPE failure reason"
- );
+ 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();
@@ -585,248 +500,119 @@ public class
TimestampedKeyValueStoreBuilderWithHeadersTest {
}
@ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void shouldReturnUnknownQueryTypeForRangeQueryOnHeadersStore(final
boolean cachingEnabled) {
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
headersStoreMaybeWithCache(cachingEnabled);
-
+ @CsvSource({"NATIVE", "ADAPTER", "IN_MEMORY"})
+ public void shouldReturnEmptyPositionInitially(final StoreType storeType) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, false);
try {
- final RangeQuery<Bytes, byte[]> query = RangeQuery.withRange(
- new Bytes("a".getBytes()),
- new Bytes("z".getBytes())
- );
- final PositionBound positionBound = PositionBound.unbounded();
- final QueryConfig config = new QueryConfig(false);
-
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final QueryResult<KeyValueIterator<Bytes, byte[]>> result =
wrapped.query(query, positionBound, config);
-
- // Verify: Headers store currently returns UNKNOWN_QUERY_TYPE
- assertFalse(result.isSuccess(), "Expected query to fail with
unknown query type");
- assertEquals(
- FailureReason.UNKNOWN_QUERY_TYPE,
- result.getFailureReason(),
- "Expected UNKNOWN_QUERY_TYPE failure reason"
- );
- assertNotNull(result.getPosition(), "Expected position to be set");
+ final Position position = ((WrappedStateStore)
store).wrapped().getPosition();
+ assertNotNull(position, "Expected non-null position");
+ assertTrue(position.isEmpty(), "Expected position to be empty
initially");
} finally {
store.close();
}
}
- @Test
- public void shouldCollectExecutionInfoForQueryOnHeadersStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStoreWithHeaders("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ @ParameterizedTest
+ @CsvSource({"NATIVE", "ADAPTER"})
+ public void shouldCollectExecutionInfoWhenRequested(final StoreType
storeType) {
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(storeType, false);
try {
- final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(new
Bytes("test-key".getBytes()));
- final PositionBound positionBound = PositionBound.unbounded();
- final QueryConfig config = new QueryConfig(true); // Enable
execution info
-
final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final QueryResult<byte[]> result = wrapped.query(query,
positionBound, config);
+ final QueryResult<byte[]> result = wrapped.query(
+ KeyQuery.withKey(new Bytes("k".getBytes())),
PositionBound.unbounded(), new QueryConfig(true));
- // Verify: Execution info was collected
- assertFalse(result.getExecutionInfo().isEmpty(), "Expected
execution info to be collected");
- assertTrue(
- result.getExecutionInfo().get(0).contains("Handled in"),
- "Expected execution info to contain handling information"
- );
- assertTrue(
-
result.getExecutionInfo().get(0).contains(RocksDBTimestampedStoreWithHeaders.class.getName()),
- "Expected execution info to mention the class name"
- );
+ final String executionInfo = String.join("\n",
result.getExecutionInfo());
+ assertFalse(executionInfo.isEmpty(), "Expected execution info to
be collected");
+ assertTrue(executionInfo.contains("Handled in"), "Expected
execution info to contain handling information");
+ final String expectedClass = storeType == StoreType.NATIVE
+ ? RocksDBTimestampedStoreWithHeaders.class.getName()
+ : TimestampedToHeadersStoreAdapter.class.getName();
+ assertTrue(executionInfo.contains(expectedClass), "Expected
execution info to mention " + expectedClass);
} finally {
store.close();
}
}
@Test
- public void shouldHandleKeyQueryOnAdaptedTimestampedStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStore("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ public void
shouldCollectExecutionInfoForTimestampedKeyWithHeadersQueryWhenRequested() {
+ // The typed query, run through the metered handler with execution
info enabled, must carry both
+ // the wrapped store's entry and the metered handler's entry -- on the
success path and on the
+ // negative-timestamp failure path (which builds a fresh failure
result).
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
try {
- // Put data into the store (headers will be discarded when adapted
to timestamped store)
- final Headers headers = new RecordHeaders();
- headers.add("adapter", "test".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("adapter-value",
55555L, headers));
+ store.put("ok", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+ store.put("bad", ValueTimestampHeaders.make("v", -1L,
headersWith("h", "x")));
- // Verify adapter is used for legacy timestamped store
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(TimestampedToHeadersStoreAdapter.class, wrapped,
- "Expected TimestampedToHeadersStoreAdapter for legacy
timestamped store");
+ final QueryResult<ReadOnlyRecord<String, String>> ok =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("ok"),
PositionBound.unbounded(), new QueryConfig(true));
+ final QueryResult<ReadOnlyRecord<String, String>> bad =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("bad"),
PositionBound.unbounded(), new QueryConfig(true));
- // Query at typed level - KeyQuery should return just the value
- final KeyQuery<String, String> query =
KeyQuery.withKey("test-key");
- final QueryResult<String> result = store.query(query,
PositionBound.unbounded(), new QueryConfig(false));
+ assertTrue(ok.isSuccess());
+ assertFalse(bad.isSuccess());
+ assertEquals(FailureReason.STORE_EXCEPTION,
bad.getFailureReason());
- // Verify IQv2 query result
- // Adapter delegates to RocksDBTimestampedStore which supports
IQv2 through RocksDBStore
- assertTrue(result.isSuccess(),
- "Expected query to succeed since RocksDBTimestampedStore
supports IQv2");
- assertNotNull(result.getPosition(), "Expected position to be set");
- assertInstanceOf(String.class, result.getResult());
- assertEquals("adapter-value", result.getResult(), "KeyQuery should
return just the value");
+ final String okInfo = String.join("\n", ok.getExecutionInfo());
+ final String badInfo = String.join("\n", bad.getExecutionInfo());
+ assertTrue(
+
okInfo.contains(RocksDBTimestampedStoreWithHeaders.class.getName())
+ &&
okInfo.contains(MeteredTimestampedKeyValueStoreWithHeaders.class.getName()),
+ "success execution info missing an entry: " + okInfo);
+ assertTrue(
+
badInfo.contains(RocksDBTimestampedStoreWithHeaders.class.getName())
+ &&
badInfo.contains(MeteredTimestampedKeyValueStoreWithHeaders.class.getName()),
+ "failure execution info missing an entry (wrapped-store entry
must be preserved): " + badInfo);
} finally {
store.close();
}
}
@Test
- public void shouldHandleTimestampedKeyQueryOnAdaptedTimestampedStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStore("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
-
+ public void
shouldNotCollectExecutionInfoForTimestampedKeyWithHeadersQueryWhenNotRequested()
{
+ // With execution info disabled, neither the success path nor the
negative-timestamp failure path
+ // may collect any (the failure path must not copy execution info
unconditionally).
+ final TimestampedKeyValueStoreWithHeaders<String, String> store =
buildAndInitStore(StoreType.NATIVE, false);
try {
- // Put data into the store (headers will be discarded when adapted
to timestamped store)
- final Headers headers = new RecordHeaders();
- headers.add("adapter", "test".getBytes());
- store.put("test-key", ValueTimestampHeaders.make("adapter-value",
55555L, headers));
-
- // Verify adapter is used for legacy timestamped store
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- assertInstanceOf(TimestampedToHeadersStoreAdapter.class, wrapped,
- "Expected TimestampedToHeadersStoreAdapter for legacy
timestamped store");
-
- // Query at typed level - TimestampedKeyQuery should return value
+ timestamp
- final TimestampedKeyQuery<String, String> query =
TimestampedKeyQuery.withKey("test-key");
- final QueryResult<ValueAndTimestamp<String>> result =
store.query(query, PositionBound.unbounded(), new QueryConfig(false));
-
- // Verify IQv2 query result
- // Adapter delegates to RocksDBTimestampedStore which supports
IQv2 through RocksDBStore
- assertTrue(result.isSuccess(),
- "Expected query to succeed since RocksDBTimestampedStore
supports IQv2");
- assertNotNull(result.getPosition(), "Expected position to be set");
- assertNotNull(result.getResult(), "Expected non-null result");
- assertInstanceOf(ValueAndTimestamp.class, result.getResult());
- assertEquals("adapter-value", result.getResult().value(),
"TimestampedKeyQuery should return the value");
- assertEquals(55555L, result.getResult().timestamp(),
"TimestampedKeyQuery should return the timestamp");
+ store.put("ok", ValueTimestampHeaders.make("v", 123L,
headersWith("h", "x")));
+ store.put("bad", ValueTimestampHeaders.make("v", -1L,
headersWith("h", "x")));
+
+ final QueryResult<ReadOnlyRecord<String, String>> ok =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("ok"),
PositionBound.unbounded(), new QueryConfig(false));
+ final QueryResult<ReadOnlyRecord<String, String>> bad =
+ store.query(TimestampedKeyWithHeadersQuery.withKey("bad"),
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(ok.isSuccess());
+ assertFalse(bad.isSuccess());
+ assertTrue(ok.getExecutionInfo().isEmpty(), "Expected no execution
info on success: " + ok.getExecutionInfo());
+ assertTrue(bad.getExecutionInfo().isEmpty(), "Expected no
execution info on failure: " + bad.getExecutionInfo());
} finally {
store.close();
}
}
- @Test
- public void shouldCollectExecutionInfoForQueryOnAdaptedTimestampedStore() {
- when(supplier.name()).thenReturn("test-store");
- when(supplier.metricsScope()).thenReturn("metricScope");
- when(supplier.get()).thenReturn(new
RocksDBTimestampedStore("test-store", "metrics-scope"));
-
- builder = new TimestampedKeyValueStoreBuilderWithHeaders<>(
- supplier,
- Serdes.String(),
- Serdes.String(),
- new MockTime()
- );
-
- final TimestampedKeyValueStoreWithHeaders<String, String> store =
builder
- .withLoggingDisabled()
- .withCachingDisabled()
- .build();
-
- final File dir = TestUtils.tempDirectory();
- final Properties props = StreamsTestUtils.getStreamsConfig();
- final InternalMockProcessorContext<String, String> context = new
InternalMockProcessorContext<>(
- dir,
- Serdes.String(),
- Serdes.String(),
- new StreamsConfig(props)
- );
- store.init(context, store);
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void
shouldReturnIdenticalResultsForNativeAndAdapterBuiltStores(final boolean
cachingEnabled) {
+ // The existing (header-stripped) query types must return identical
results on both build paths.
+ final ValueTimestampHeaders<String> value =
ValueTimestampHeaders.make("v", 123L, headersWith("h", "x"));
+ final TimestampedKeyValueStoreWithHeaders<String, String> nativeStore
= buildAndInitStore(StoreType.NATIVE, cachingEnabled);
+ final TimestampedKeyValueStoreWithHeaders<String, String> adapterStore
= buildAndInitStore(StoreType.ADAPTER, cachingEnabled);
try {
- final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(new
Bytes("test-key".getBytes()));
- final PositionBound positionBound = PositionBound.unbounded();
- final QueryConfig config = new QueryConfig(true); // Enable
execution info
+ nativeStore.put("k", value);
+ adapterStore.put("k", value);
- final StateStore wrapped = ((WrappedStateStore) store).wrapped();
- final QueryResult<byte[]> result = wrapped.query(query,
positionBound, config);
-
- // Verify: Execution info includes both adapter and underlying
store
- assertFalse(result.getExecutionInfo().isEmpty(), "Expected
execution info to be collected");
-
- final String executionInfo = String.join("\n",
result.getExecutionInfo());
- assertTrue(
- executionInfo.contains("Handled in"),
- "Expected execution info to contain handling information"
- );
- // Should mention the adapter class
- assertTrue(
-
executionInfo.contains(TimestampedToHeadersStoreAdapter.class.getName()),
- "Expected execution info to mention the adapter class"
- );
+ assertEquals(
+ nativeStore.query(KeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false)).getResult(),
+ adapterStore.query(KeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false)).getResult(),
+ "KeyQuery results should be identical across native and
adapter build paths");
+ assertEquals(
+ nativeStore.query(TimestampedKeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false)).getResult(),
+ adapterStore.query(TimestampedKeyQuery.withKey("k"),
PositionBound.unbounded(), new QueryConfig(false)).getResult(),
+ "TimestampedKeyQuery results should be identical across native
and adapter build paths");
} finally {
- store.close();
+ nativeStore.close();
+ adapterStore.close();
}
}
}
