You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by fr...@apache.org on 2026/09/22 17:22:30 UTC
(kafka) branch trunk updated: KAFKA-16158: Cleanup usage of `TimestampedBytesStore` interface (#23233)
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 8e65883c125 KAFKA-16158: Cleanup usage of `TimestampedBytesStore` interface (#23233)
8e65883c125 is described below
commit 8e65883c1255f93361bbf29efa8425e30b40c5ec
Author: gabriellafu <10...@users.noreply.github.com>
AuthorDate: Tue Sep 22 13:22:22 2026 -0400
KAFKA-16158: Cleanup usage of `TimestampedBytesStore` interface (#23233)
1. `KeyValueToTimestampedKeyValueByteStoreAdapter` now implements
`TimestampedBytesStore`, fixing the inconsistency where it stored
`<timestamp><value>` data without advertising the interface.
2. Reordered `StateManagerUtil.converterForStore` to check the wrapped
chain for adapters first, returning `identity()` for the KV adapter
before the `isHeadersAware`/`isTimestamped` checks.
3. Removed the now-redundant `StoreQueryUtils.isAdapter()` helper and
its two call sites (`CachingKeyValueStore`,
`StoreQueryUtils.deserializeValue`), since
`WrappedStateStore.isTimestamped()` now returns `true` for the adapter.
4. Rewrote `StateManagerUtilConverterTest` to use real store-builder
chains instead of mocks, giving genuine round-trip regression coverage
for KAFKA-16141 (in-memory retains timestamp `42`, persistent-plain uses
dummy `-1`).
5. added tests for window store round trip, iqv2 cache read path, and
deserializeValue guard
Reviewers: Alieh Saeedi <as...@confluent.io>, TengYao Chi
<fr...@apache.org>
---
.../processor/internals/StateManagerUtil.java | 46 +++--
.../state/internals/CachingKeyValueStore.java | 21 +-
.../internals/CachingKeyValueStoreWithHeaders.java | 2 +-
.../state/internals/KeyValueStoreBuilder.java | 2 +-
...ValueToTimestampedKeyValueByteStoreAdapter.java | 3 +-
.../state/internals/ListValueStoreBuilder.java | 2 +-
.../streams/state/internals/StoreQueryUtils.java | 13 +-
.../internals/TimestampedKeyValueStoreBuilder.java | 7 +-
.../internals/StateManagerUtilConverterTest.java | 228 ++++++++++++++++-----
.../CachingInMemoryKeyValueStoreTest.java | 6 +-
.../internals/CachingKeyValueStoreAdapterTest.java | 100 +++++++++
.../state/internals/MeteredKeyValueStoreTest.java | 5 +-
.../state/internals/StoreQueryUtilsTest.java | 26 +++
13 files changed, 357 insertions(+), 104 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
index 2612dd0373b..cac833fba18 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java
@@ -27,11 +27,14 @@ import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.Task.TaskType;
import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.internals.KeyValueToTimestampedKeyValueByteStoreAdapter;
import org.apache.kafka.streams.state.internals.PlainToHeadersStoreAdapter;
import org.apache.kafka.streams.state.internals.PlainToHeadersWindowStoreAdapter;
import org.apache.kafka.streams.state.internals.RecordConverter;
+import org.apache.kafka.streams.state.internals.SessionToHeadersStoreAdapter;
import org.apache.kafka.streams.state.internals.TimestampedToHeadersStoreAdapter;
import org.apache.kafka.streams.state.internals.TimestampedToHeadersWindowStoreAdapter;
+import org.apache.kafka.streams.state.internals.WindowToTimestampedWindowByteStoreAdapter;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
import org.slf4j.Logger;
@@ -59,26 +62,27 @@ final class StateManagerUtil {
private StateManagerUtil() {}
static RecordConverter converterForStore(final StateStore store) {
- // First check if the top-level store implements HeadersBytesStore or TimestampedBytesStore
- if (isHeadersAware(store)) {
- if (store instanceof SessionStore) {
- return rawValueToSessionHeadersValue();
- }
- return rawValueToHeadersValue();
- } else if (isTimestamped(store) && !isVersioned(store)) {
- // should not prepend timestamp when restoring records for versioned store, as
- // timestamp is used separately during put() process for restore of versioned stores
- return rawValueToTimestampedValue();
- }
-
- // If top-level check didn't find the type, unwrap to find adapters
- // This handles persistent stores that use adapters
+ // Restore bypasses adapters and writes directly into the inner store, so the converter must
+ // match the inner store's binary format, not the format the adapter advertises to the outer
+ // store chain. Thus, check for adapters first.
+ //
+ // This loop enumerates the byte-translating adapters and maps each to its INNER store's
+ // format. Wrappers that merely advertise a format without translating bytes (e.g. the
+ // in-memory timestamped markers) are intentionally NOT listed here — they are resolved by
+ // the isTimestamped()/isHeadersAware() fallback below. Note that
+ // WindowToTimestampedWindowByteStoreAdapter deliberately does not implement
+ // TimestampedBytesStore: marking it without also converting in its query() would break the
+ // IQ read path.
StateStore current = store;
while (current != null) {
if (current instanceof TimestampedToHeadersStoreAdapter || current instanceof TimestampedToHeadersWindowStoreAdapter) {
// Adapter wraps a timestamped store, so restore in timestamped format
return rawValueToTimestampedValue();
- } else if (current instanceof PlainToHeadersStoreAdapter || current instanceof PlainToHeadersWindowStoreAdapter) {
+ } else if (current instanceof PlainToHeadersStoreAdapter
+ || current instanceof PlainToHeadersWindowStoreAdapter
+ || current instanceof SessionToHeadersStoreAdapter
+ || current instanceof KeyValueToTimestampedKeyValueByteStoreAdapter
+ || current instanceof WindowToTimestampedWindowByteStoreAdapter) {
// Adapter wraps a plain store, so restore in plain format
return identity();
}
@@ -92,6 +96,18 @@ final class StateManagerUtil {
current = ((WrappedStateStore<?, ?, ?>) current).wrapped();
}
+ // No adapter found: the inner store's binary format is whatever the store chain advertises
+ if (isHeadersAware(store)) {
+ if (store instanceof SessionStore) {
+ return rawValueToSessionHeadersValue();
+ }
+ return rawValueToHeadersValue();
+ } else if (isTimestamped(store) && !isVersioned(store)) {
+ // should not prepend timestamp when restoring records for versioned store, as
+ // timestamp is used separately during put() process for restore of versioned stores
+ return rawValueToTimestampedValue();
+ }
+
// Default to identity if no special handling needed
return identity();
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
index 1a1edf71692..a82d6d8ea36 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
@@ -68,7 +68,6 @@ public class CachingKeyValueStore
private Thread streamThread;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final Position position;
- private final CacheType cacheType;
@FunctionalInterface
public interface CacheQueryHandler {
@@ -92,10 +91,9 @@ public class CachingKeyValueStore
);
- CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying, final CacheType cacheType) {
+ CachingKeyValueStore(final KeyValueStore<Bytes, byte[]> underlying) {
super(underlying);
position = Position.emptyPosition();
- this.cacheType = cacheType;
}
@Override
@@ -188,13 +186,10 @@ public class CachingKeyValueStore
if (internalContext.cache() != null) {
final LRUCacheEntry lruCacheEntry = internalContext.cache().get(cacheName, key);
if (lruCacheEntry != null) {
- final byte[] rawValue;
- if (cacheType == CacheType.TIMESTAMPED_KEY_VALUE_STORE && !WrappedStateStore.isTimestamped(wrapped()) && !StoreQueryUtils.isAdapter(wrapped())) {
- rawValue = ValueAndTimestampDeserializer.rawValue(lruCacheEntry.value());
- } else {
- rawValue = lruCacheEntry.value();
- }
- result = (QueryResult<R>) QueryResult.forResult(rawValue);
+ // the cache holds the same binary format as the wrapped store: every store below a
+ // timestamped cache implements TimestampedBytesStore (natively, via the KV adapter,
+ // or via the in-memory marker), so no format conversion is needed here
+ result = (QueryResult<R>) QueryResult.forResult(lruCacheEntry.value());
}
}
@@ -564,10 +559,4 @@ public class CachingKeyValueStore
lock.writeLock().unlock();
}
}
-
- public enum CacheType {
- KEY_VALUE_STORE,
- TIMESTAMPED_KEY_VALUE_STORE,
- TIMESTAMPED_KEY_VALUE_STORE_WITH_HEADERS
- }
}
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 f2609803ed7..6e01d2a7063 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
@@ -32,6 +32,6 @@ import org.apache.kafka.streams.state.KeyValueStore;
public class CachingKeyValueStoreWithHeaders extends CachingKeyValueStore {
CachingKeyValueStoreWithHeaders(final KeyValueStore<Bytes, byte[]> underlying) {
- super(underlying, CacheType.TIMESTAMPED_KEY_VALUE_STORE_WITH_HEADERS);
+ super(underlying);
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilder.java
index 0cfbe2d575a..7888316bacf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueStoreBuilder.java
@@ -52,7 +52,7 @@ public class KeyValueStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, KeyVa
if (!enableCaching) {
return inner;
}
- return new CachingKeyValueStore(inner, CachingKeyValueStore.CacheType.KEY_VALUE_STORE);
+ return new CachingKeyValueStore(inner);
}
private KeyValueStore<Bytes, byte[]> maybeWrapLogging(final KeyValueStore<Bytes, byte[]> inner) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
index 406823a026a..b70655ab2ce 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueToTimestampedKeyValueByteStoreAdapter.java
@@ -35,6 +35,7 @@ import org.apache.kafka.streams.query.internals.InternalQueryResultUtil;
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.TimestampedBytesStore;
import java.util.List;
import java.util.Map;
@@ -54,7 +55,7 @@ import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserial
*/
@SuppressWarnings("unchecked")
-public class KeyValueToTimestampedKeyValueByteStoreAdapter implements KeyValueStore<Bytes, byte[]> {
+public class KeyValueToTimestampedKeyValueByteStoreAdapter implements KeyValueStore<Bytes, byte[]>, TimestampedBytesStore {
final KeyValueStore<Bytes, byte[]> store;
KeyValueToTimestampedKeyValueByteStoreAdapter(final KeyValueStore<Bytes, byte[]> store) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java
index 92b4c4d96e9..34e2e8b231d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ListValueStoreBuilder.java
@@ -51,7 +51,7 @@ public class ListValueStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, KeyV
if (!enableCaching) {
return inner;
}
- return new CachingKeyValueStore(inner, CachingKeyValueStore.CacheType.KEY_VALUE_STORE);
+ return new CachingKeyValueStore(inner);
}
private KeyValueStore<Bytes, byte[]> maybeWrapLogging(final KeyValueStore<Bytes, byte[]> inner) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
index 96f641ac632..d30b6f3e19b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
@@ -435,7 +435,7 @@ public final class StoreQueryUtils {
@SuppressWarnings({"unchecked", "rawtypes", "resource"})
public static <V> Function<byte[], V> deserializeValue(final StateSerdes<?, V> serdes, final StateStore wrapped) {
final Serde<V> valueSerde = serdes.valueSerde();
- final boolean timestamped = WrappedStateStore.isTimestamped(wrapped) || isAdapter(wrapped);
+ final boolean timestamped = WrappedStateStore.isTimestamped(wrapped);
final Deserializer<V> deserializer;
if (!timestamped && valueSerde instanceof ValueAndTimestampSerde) {
final ValueAndTimestampDeserializer valueAndTimestampDeserializer =
@@ -448,17 +448,6 @@ public final class StoreQueryUtils {
return byteArray -> deserializer.deserialize(serdes.topic(), new RecordHeaders(), byteArray);
}
- @SuppressWarnings("rawtypes")
- public static boolean isAdapter(final StateStore stateStore) {
- if (stateStore instanceof KeyValueToTimestampedKeyValueByteStoreAdapter) {
- return true;
- } else if (stateStore instanceof WrappedStateStore) {
- return isAdapter(((WrappedStateStore) stateStore).wrapped());
- } else {
- return false;
- }
- }
-
@SuppressWarnings("resource")
public static <V> Function<VersionedRecord<byte[]>, VersionedRecord<V>> deserializeValue(final StateSerdes<?, V> serdes) {
final Serde<V> valueSerde = serdes.valueSerde();
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilder.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilder.java
index 28516b4c8a7..b253ce94133 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedKeyValueStoreBuilder.java
@@ -76,7 +76,7 @@ public class TimestampedKeyValueStoreBuilder<K, V>
if (!enableCaching) {
return inner;
}
- return new CachingKeyValueStore(inner, CachingKeyValueStore.CacheType.TIMESTAMPED_KEY_VALUE_STORE);
+ return new CachingKeyValueStore(inner);
}
private KeyValueStore<Bytes, byte[]> maybeWrapLogging(final KeyValueStore<Bytes, byte[]> inner) {
@@ -86,6 +86,11 @@ public class TimestampedKeyValueStoreBuilder<K, V>
return new ChangeLoggingTimestampedKeyValueBytesStore(inner);
}
+ /**
+ * Unlike persistent stores, in-memory stores have no pre-existing data in the old plain format,
+ * so the wrapped store holds the {@code <timestamp,value>} format natively and no byte-array
+ * translation is needed; this marker only advertises the format to the restore and IQ code paths.
+ */
private static final class InMemoryTimestampedKeyValueStoreMarker
extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, Bytes, byte[]>
implements KeyValueStore<Bytes, byte[]>, TimestampedBytesStore {
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java
index 34bf9979cb6..02cc8137a69 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java
@@ -16,32 +16,58 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedWindowStore;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+import org.apache.kafka.streams.state.VersionedKeyValueStore;
+import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.InMemoryKeyValueStore;
import org.apache.kafka.streams.state.internals.InMemorySessionStore;
import org.apache.kafka.streams.state.internals.InMemoryWindowStore;
-import org.apache.kafka.streams.state.internals.KeyValueToTimestampedKeyValueByteStoreAdapter;
import org.apache.kafka.streams.state.internals.MeteredSessionStoreWithHeaders;
-import org.apache.kafka.streams.state.internals.MeteredTimestampedKeyValueStore;
import org.apache.kafka.streams.state.internals.MeteredTimestampedKeyValueStoreWithHeaders;
-import org.apache.kafka.streams.state.internals.MeteredTimestampedWindowStore;
import org.apache.kafka.streams.state.internals.MeteredTimestampedWindowStoreWithHeaders;
import org.apache.kafka.streams.state.internals.PlainToHeadersStoreAdapter;
import org.apache.kafka.streams.state.internals.PlainToHeadersWindowStoreAdapter;
import org.apache.kafka.streams.state.internals.RecordConverter;
import org.apache.kafka.streams.state.internals.SessionToHeadersStoreAdapter;
+import org.apache.kafka.streams.state.internals.TimestampedKeyValueStoreBuilder;
import org.apache.kafka.streams.state.internals.TimestampedToHeadersStoreAdapter;
import org.apache.kafka.streams.state.internals.TimestampedToHeadersWindowStoreAdapter;
-import org.apache.kafka.streams.state.internals.WindowToTimestampedWindowByteStoreAdapter;
+import org.apache.kafka.streams.state.internals.TimestampedWindowStoreBuilder;
+import org.apache.kafka.streams.state.internals.WindowKeySchema;
import org.apache.kafka.streams.state.internals.WrappedStateStore;
+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.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.quality.Strictness;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
import static org.apache.kafka.streams.state.internals.RecordConverters.identity;
import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToHeadersValue;
import static org.apache.kafka.streams.state.internals.RecordConverters.rawValueToSessionHeadersValue;
@@ -55,31 +81,103 @@ import static org.mockito.Mockito.withSettings;
@MockitoSettings(strictness = Strictness.STRICT_STUBS)
public class StateManagerUtilConverterTest {
- @Test
- public void shouldReturnIdentityConverterForPlainToTimestampedPersistentKeyValueStore() {
- // persistent plain kv -> ts kv
- final WrappedStateStore<?, ?, ?> mockWrapper = mock(WrappedStateStore.class);
- final StateStore mockAdapter = mock(KeyValueToTimestampedKeyValueByteStoreAdapter.class);
+ private static final long TIMESTAMP = 42L;
+ private static final long WINDOW_START = 0L;
- doReturn(mockAdapter).when(mockWrapper).wrapped();
+ // persistent plain kv/window -> ts kv/window (via KeyValueToTimestampedKeyValueByteStoreAdapter /
+ // WindowToTimestampedWindowByteStoreAdapter): restore bypasses the adapter and writes into the
+ // plain inner store directly, so the converter must be identity(). In-memory and persistent
+ // timestamped stores hold the timestamped format natively, so the converter must prepend it.
+ @ParameterizedTest
+ @MethodSource("keyValueConverterCases")
+ public void shouldReturnConverterForTimestampedKeyValueStore(final KeyValueBytesStoreSupplier supplier,
+ final RecordConverter expectedConverter) {
+ final TimestampedKeyValueStore<String, String> store = timestampedKeyValueStore(supplier);
- final RecordConverter converter = StateManagerUtil.converterForStore(mockWrapper);
+ assertEquals(expectedConverter, StateManagerUtil.converterForStore(store));
+ }
- assertEquals(identity(), converter);
+ @ParameterizedTest
+ @MethodSource("windowConverterCases")
+ public void shouldReturnConverterForTimestampedWindowStore(final WindowBytesStoreSupplier supplier,
+ final RecordConverter expectedConverter) {
+ final StateStore store = timestampedWindowStore(supplier);
+
+ assertEquals(expectedConverter, StateManagerUtil.converterForStore(store));
}
+ // versioned stores implement VersionedBytesStore, which extends TimestampedBytesStore, so both
+ // isTimestamped() and isVersioned() are true; the isVersioned() guard in converterForStore must
+ // win so that restore does not prepend a timestamp (versioned stores handle it separately in put())
@Test
- @SuppressWarnings("unchecked")
- public void shouldReturnIdentityConverterForPlainToTimestampedInMemoryKeyValueStore() {
- // in memory kv -> ts kv (using InMemoryTimestampedKeyValueStoreMarker)
- final StateStore mockInnerStore = mock(InMemoryKeyValueStore.class);
- final WrappedStateStore<?, ?, ?> mockMarker = mock(MeteredTimestampedKeyValueStore.class);
+ public void shouldReturnIdentityConverterForVersionedKeyValueStore() {
+ final VersionedKeyValueStore<String, String> store = Stores.versionedKeyValueStoreBuilder(
+ Stores.persistentVersionedKeyValueStore("store", Duration.ofMillis(1000)),
+ Serdes.String(),
+ Serdes.String()
+ ).build();
+
+ assertEquals(identity(), StateManagerUtil.converterForStore(store));
+ }
- doReturn(mockInnerStore).when(mockMarker).wrapped();
+ // regression test for KAFKA-16141: restore bypasses the adapter, so restored records must not get
+ // a timestamp prepended even though the adapter advertises the timestamped format. The plain inner
+ // store cannot retain the record timestamp, so reads surface the dummy `-1`; in-memory and
+ // persistent timestamped stores retain the real timestamp.
+ @ParameterizedTest
+ @MethodSource("keyValueRestoreCases")
+ public void shouldRestoreTimestampedKeyValueStore(final KeyValueBytesStoreSupplier supplier,
+ final long expectedTimestamp) {
+ final TimestampedKeyValueStore<String, String> store = timestampedKeyValueStore(supplier);
- final RecordConverter converter = StateManagerUtil.converterForStore(mockMarker);
+ final ValueAndTimestamp<String> restored = restoreAndGet(store);
- assertEquals(identity(), converter);
+ assertEquals("value", restored.value());
+ assertEquals(expectedTimestamp, restored.timestamp());
+ }
+
+ @ParameterizedTest
+ @MethodSource("windowRestoreCases")
+ public void shouldRestoreTimestampedWindowStore(final WindowBytesStoreSupplier supplier,
+ final long expectedTimestamp) {
+ final TimestampedWindowStore<String, String> store = timestampedWindowStore(supplier);
+
+ final ValueAndTimestamp<String> restored = restoreAndGet(store);
+
+ assertEquals("value", restored.value());
+ assertEquals(expectedTimestamp, restored.timestamp());
+ }
+
+ private static Stream<Arguments> keyValueConverterCases() {
+ return Stream.of(
+ Arguments.of(Stores.persistentKeyValueStore("store"), identity()),
+ Arguments.of(Stores.inMemoryKeyValueStore("store"), rawValueToTimestampedValue()),
+ Arguments.of(Stores.persistentTimestampedKeyValueStore("store"), rawValueToTimestampedValue())
+ );
+ }
+
+ private static Stream<Arguments> windowConverterCases() {
+ return Stream.of(
+ Arguments.of(Stores.persistentWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), identity()),
+ Arguments.of(Stores.inMemoryWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), rawValueToTimestampedValue()),
+ Arguments.of(Stores.persistentTimestampedWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), rawValueToTimestampedValue())
+ );
+ }
+
+ private static Stream<Arguments> keyValueRestoreCases() {
+ return Stream.of(
+ Arguments.of(Stores.persistentKeyValueStore("store"), -1L),
+ Arguments.of(Stores.inMemoryKeyValueStore("store"), TIMESTAMP),
+ Arguments.of(Stores.persistentTimestampedKeyValueStore("store"), TIMESTAMP)
+ );
+ }
+
+ private static Stream<Arguments> windowRestoreCases() {
+ return Stream.of(
+ Arguments.of(Stores.persistentWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), -1L),
+ Arguments.of(Stores.inMemoryWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), TIMESTAMP),
+ Arguments.of(Stores.persistentTimestampedWindowStore("store", Duration.ofMillis(1000), Duration.ofMillis(100), false), TIMESTAMP)
+ );
}
@Test
@@ -123,33 +221,6 @@ public class StateManagerUtilConverterTest {
assertEquals(rawValueToTimestampedValue(), converter);
}
- @Test
- public void shouldReturnIdentityConverterForPlainToTimestampedPersistentWindowStore() {
- // persistent plain window -> ts window
- final WrappedStateStore<?, ?, ?> mockWrapper = mock(WrappedStateStore.class);
- final StateStore mockAdapter = mock(WindowToTimestampedWindowByteStoreAdapter.class);
-
- doReturn(mockAdapter).when(mockWrapper).wrapped();
-
- final RecordConverter converter = StateManagerUtil.converterForStore(mockWrapper);
-
- assertEquals(identity(), converter);
- }
-
- @Test
- @SuppressWarnings("unchecked")
- public void shouldReturnIdentityConverterForPlainToTimestampedInMemoryWindowStore() {
- // in memory window -> ts window (using InMemoryTimestampedWindowStoreMarker)
- final StateStore mockInnerStore = mock(InMemoryKeyValueStore.class);
- final WrappedStateStore<?, ?, ?> mockMarker = mock(MeteredTimestampedWindowStore.class);
-
- doReturn(mockInnerStore).when(mockMarker).wrapped();
-
- final RecordConverter converter = StateManagerUtil.converterForStore(mockMarker);
-
- assertEquals(identity(), converter);
- }
-
@Test
public void shouldReturnIdentityConverterForPlainToHeadersPersistentWindowStore() {
// persistent plain window -> headers window
@@ -206,7 +277,7 @@ public class StateManagerUtilConverterTest {
@Test
@SuppressWarnings("unchecked")
- public void shouldReturnIdentityConverterForPlainToHeadersInMemorySessionStore() {
+ public void shouldReturnSessionHeadersConverterForPlainToHeadersInMemorySessionStore() {
// in memory session -> headers session (using InMemorySessionStoreWithHeadersMarker)
final StateStore mockInnerStore = mock(InMemorySessionStore.class, withSettings().extraInterfaces(HeadersBytesStore.class));
final WrappedStateStore<?, ?, ?> mockMarker = mock(MeteredSessionStoreWithHeaders.class);
@@ -218,4 +289,63 @@ public class StateManagerUtilConverterTest {
assertEquals(rawValueToSessionHeadersValue(), converter);
}
-}
\ No newline at end of file
+ private static TimestampedKeyValueStore<String, String> timestampedKeyValueStore(final KeyValueBytesStoreSupplier supplier) {
+ return new TimestampedKeyValueStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), Time.SYSTEM).build();
+ }
+
+ private static TimestampedWindowStore<String, String> timestampedWindowStore(final WindowBytesStoreSupplier supplier) {
+ return new TimestampedWindowStoreBuilder<>(supplier, Serdes.String(), Serdes.String(), Time.SYSTEM).build();
+ }
+
+ private static ValueAndTimestamp<String> restoreAndGet(final TimestampedWindowStore<String, String> store) {
+ // window changelog keys carry the window start (here WINDOW_START) in their binary encoding
+ final byte[] key = WindowKeySchema.toStoreKeyBinary(
+ Bytes.wrap("key".getBytes(StandardCharsets.UTF_8)), WINDOW_START, 0).get();
+ return restoreAndRead(store, key, () -> store.fetch("key", WINDOW_START));
+ }
+
+ private static ValueAndTimestamp<String> restoreAndGet(final TimestampedKeyValueStore<String, String> store) {
+ return restoreAndRead(store, "key".getBytes(StandardCharsets.UTF_8), () -> store.get("key"));
+ }
+
+ /**
+ * Feeds a changelog-format record (plain value, timestamp in the record timestamp field) through
+ * the converter and the store's registered restore callback, mirroring the restore code path,
+ * then reads the restored value back via {@code read}.
+ */
+ private static ValueAndTimestamp<String> restoreAndRead(final StateStore store,
+ final byte[] key,
+ final Supplier<ValueAndTimestamp<String>> read) {
+ final InternalMockProcessorContext<?, ?> context = new InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(StreamsTestUtils.getStreamsConfig())
+ );
+ store.init(context, store);
+ try {
+ final byte[] plainValue = "value".getBytes(StandardCharsets.UTF_8);
+ final ConsumerRecord<byte[], byte[]> changelogRecord = new ConsumerRecord<>(
+ "changelog",
+ 0,
+ 0L,
+ TIMESTAMP,
+ TimestampType.CREATE_TIME,
+ key.length,
+ plainValue.length,
+ key,
+ plainValue,
+ new RecordHeaders(),
+ Optional.empty()
+ );
+
+ final RecordConverter converter = StateManagerUtil.converterForStore(store);
+ context.restoreWithHeaders(store.name(), List.of(converter.convert(changelogRecord)));
+
+ return read.get();
+ } finally {
+ store.close();
+ }
+ }
+
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingInMemoryKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingInMemoryKeyValueStoreTest.java
index ed8abdcd137..d3682250ece 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingInMemoryKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingInMemoryKeyValueStoreTest.java
@@ -87,7 +87,7 @@ public class CachingInMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest
final String storeName = "store";
underlyingStore = new InMemoryKeyValueStore(storeName);
cacheFlushListener = new CacheFlushListenerStub<>(new StringDeserializer(), new StringDeserializer());
- store = new CachingKeyValueStore(underlyingStore, CachingKeyValueStore.CacheType.KEY_VALUE_STORE);
+ store = new CachingKeyValueStore(underlyingStore);
store.setFlushListener(cacheFlushListener, false);
cache = new ThreadCache(new LogContext("testCache "), maxCacheSizeBytes, new MockStreamsMetrics(new Metrics()));
context = new InternalMockProcessorContext<>(null, null, null, null, cache);
@@ -112,7 +112,7 @@ public class CachingInMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest
@Test
public void shouldDelegateInit() {
final KeyValueStore<Bytes, byte[]> inner = mock(InMemoryKeyValueStore.class);
- final CachingKeyValueStore outer = new CachingKeyValueStore(inner, CachingKeyValueStore.CacheType.KEY_VALUE_STORE);
+ final CachingKeyValueStore outer = new CachingKeyValueStore(inner);
when(inner.name()).thenReturn("store");
outer.init(context, outer);
verify(inner).init(context, outer);
@@ -181,7 +181,7 @@ public class CachingInMemoryKeyValueStoreTest extends AbstractKeyValueStoreTest
private void setUpCloseTests() {
underlyingStore = mock(KeyValueStore.class);
when(underlyingStore.name()).thenReturn("store-name");
- store = new CachingKeyValueStore(underlyingStore, CachingKeyValueStore.CacheType.TIMESTAMPED_KEY_VALUE_STORE);
+ store = new CachingKeyValueStore(underlyingStore);
cache = mock(ThreadCache.class);
context = new InternalMockProcessorContext<>(TestUtils.tempDirectory(), null, null, null, cache);
context.setRecordContext(new ProcessorRecordContext(10, 0, 0, TOPIC, new RecordHeaders()));
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreAdapterTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreAdapterTest.java
new file mode 100644
index 00000000000..6e2ba22f057
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreAdapterTest.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.state.internals;
+
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.common.utils.internals.LogContext;
+import org.apache.kafka.streams.processor.internals.MockStreamsMetrics;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.query.KeyQuery;
+import org.apache.kafka.streams.query.PositionBound;
+import org.apache.kafka.streams.query.QueryConfig;
+import org.apache.kafka.streams.query.QueryResult;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.TestUtils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+
+import static org.apache.kafka.streams.state.TimestampedBytesStore.convertToTimestampedFormat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+
+/**
+ * IQv2 read path for a TimestampedKeyValueStore backed by a plain persistent store (via
+ * {@link KeyValueToTimestampedKeyValueByteStoreAdapter}): a cached entry is returned in
+ * timestamped format as-is, while a cache-bypassing read surfaces the adapter's dummy `-1`
+ * timestamp because the inner store never held one.
+ */
+public class CachingKeyValueStoreAdapterTest {
+
+ private static final Bytes KEY = Bytes.wrap("key".getBytes(StandardCharsets.UTF_8));
+ private static final byte[] PLAIN_VALUE = "value".getBytes(StandardCharsets.UTF_8);
+ private static final byte[] VALUE_WITH_TIMESTAMP = ByteBuffer
+ .allocate(8 + PLAIN_VALUE.length)
+ .putLong(42L)
+ .put(PLAIN_VALUE)
+ .array();
+
+ private CachingKeyValueStore store;
+
+ @BeforeEach
+ public void setUp() {
+ store = new CachingKeyValueStore(
+ new KeyValueToTimestampedKeyValueByteStoreAdapter(new RocksDBStore("store", "test-scope")));
+ final ThreadCache cache = new ThreadCache(new LogContext("test "), 1_000_000, new MockStreamsMetrics(new Metrics()));
+ final InternalMockProcessorContext<?, ?> context =
+ new InternalMockProcessorContext<>(TestUtils.tempDirectory(), null, null, null, cache);
+ context.setRecordContext(new ProcessorRecordContext(10, 0, 0, "topic", new RecordHeaders()));
+ store.init(context, store);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ store.close();
+ }
+
+ @Test
+ public void shouldReturnTimestampedBytesFromCacheHit() {
+ store.put(KEY, VALUE_WITH_TIMESTAMP);
+
+ final QueryResult<byte[]> result =
+ store.query(KeyQuery.withKey(KEY), PositionBound.unbounded(), new QueryConfig(false));
+
+ // the cache holds the timestamped bytes and must not strip the timestamp, since the
+ // adapter advertises the timestamped format
+ assertArrayEquals(VALUE_WITH_TIMESTAMP, result.getResult());
+ }
+
+ @Test
+ public void shouldReturnDummyTimestampFromAdapterOnCacheBypass() {
+ store.put(KEY, VALUE_WITH_TIMESTAMP);
+ store.commit(Map.of());
+
+ final QueryResult<byte[]> result =
+ store.query(KeyQuery.<Bytes, byte[]>withKey(KEY).skipCache(), PositionBound.unbounded(), new QueryConfig(false));
+
+ // the flush through the adapter stripped the timestamp, so the read fabricates `-1`
+ assertArrayEquals(convertToTimestampedFormat(PLAIN_VALUE), result.getResult());
+ }
+}
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java
index 26c41492669..5c151011b5a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStoreTest.java
@@ -385,10 +385,7 @@ public class MeteredKeyValueStoreTest {
);
final InMemoryKeyValueStore innerStore = new InMemoryKeyValueStore(STORE_NAME);
- final CachingKeyValueStore cachingStore = new CachingKeyValueStore(
- innerStore,
- CachingKeyValueStore.CacheType.KEY_VALUE_STORE
- );
+ final CachingKeyValueStore cachingStore = new CachingKeyValueStore(innerStore);
final MeteredKeyValueStore<String, String> meteredStore = new MeteredKeyValueStore<>(
cachingStore,
STORE_TYPE,
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreQueryUtilsTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreQueryUtilsTest.java
index 76c5d76a2f5..4f75f47aab0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreQueryUtilsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StoreQueryUtilsTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.processor.StateStoreContext;
import org.apache.kafka.streams.processor.TaskId;
@@ -34,12 +35,15 @@ import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.ReadOnlySessionStore;
import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.StateSerdes;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.WindowStore;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.time.Instant;
+import java.util.function.Function;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -97,6 +101,28 @@ public class StoreQueryUtilsTest {
queryResult.getFailureMessage());
}
+ @Test
+ public void shouldDeserializeInTimestampedFormatForAdapterBackedStore() {
+ // Dropping StoreQueryUtils.isAdapter() relies on KeyValueToTimestampedKeyValueByteStoreAdapter
+ // now implementing TimestampedBytesStore: isTimestamped() detects it, so IQ deserialization
+ // keeps the <timestamp,value> format instead of stripping the timestamp off as inner bytes.
+ final StateSerdes<String, ValueAndTimestamp<String>> serdes =
+ new StateSerdes<>("topic", Serdes.String(), new ValueAndTimestampSerde<>(Serdes.String()));
+ // the adapter only ever wraps a persistent store; it is never initialized here, since
+ // deserializeValue() only inspects the store type via isTimestamped()
+ final KeyValueToTimestampedKeyValueByteStoreAdapter adapterStore =
+ new KeyValueToTimestampedKeyValueByteStoreAdapter(new RocksDBStore("store", "rocksdb-state"));
+
+ final Function<byte[], ValueAndTimestamp<String>> deserializer =
+ StoreQueryUtils.deserializeValue(serdes, adapterStore);
+
+ final ValueAndTimestamp<String> result =
+ deserializer.apply(serdes.rawValue(ValueAndTimestamp.make("value", 42L)));
+
+ assertEquals("value", result.value());
+ assertEquals(42L, result.timestamp());
+ }
+
@Test
public void shouldPassIsolationLevelToReadOnlyForRangeQuery() {
@SuppressWarnings("unchecked") final KeyValueStore<Bytes, byte[]> store =