frankvicky commented on code in PR #23233:
URL: https://github.com/apache/kafka/pull/23233#discussion_r3906323162
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -59,26 +61,26 @@ 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.
Review Comment:
The reasoning here makes sense — marking
`WindowToTimestampedWindowByteStoreAdapter` without also converting in its
`query()` would break window IQ. But this leaves the interface inconsistency
that KAFKA-16158 is about half-resolved on the window side (`fetch()` returns
timestamped bytes while `query()` returns plain bytes). Could we file a
follow-up JIRA to fix the window adapter's `query()` to convert and then add
the marker, and reference it from the ticket? I'd rather have this recorded as
a ticket than only in a code comment.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -59,26 +61,26 @@ 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
KeyValueToTimestampedKeyValueByteStoreAdapter
+ || current instanceof
WindowToTimestampedWindowByteStoreAdapter) {
Review Comment:
Should we also list `SessionToHeadersStoreAdapter` in this branch? A
persistent session-headers chain currently gets the right result only
implicitly: the loop finds no match, the `isHeadersAware()` recursion is
blocked by the adapter (it is neither `HeadersBytesStore` nor a
`WrappedStateStore`), and we fall through to the trailing `identity()`. The
behavior is correct, but since this loop is now the explicit enumeration of
"adapter -> inner-store format", adding it here would make the session case
explicit and defend against future changes to the fall-through logic.
##########
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", "rocksdb-state")));
Review Comment:
nit: the second constructor argument is the metrics scope, so
`"rocksdb-state"` is a bit misleading — something like `"test-scope"` would be
clearer.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java:
##########
@@ -55,31 +81,103 @@
@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) {
Review Comment:
nit: the continuation parameter is indented one column too far (should align
with the first parameter). Same in
`shouldReturnConverterForTimestampedWindowStore`,
`shouldRestoreTimestampedKeyValueStore`, and
`shouldRestoreTimestampedWindowStore`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java:
##########
@@ -59,26 +61,26 @@ 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.
Review Comment:
@aliehsaeedii make sense?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilConverterTest.java:
##########
@@ -218,4 +289,63 @@ public void
shouldReturnIdentityConverterForPlainToHeadersInMemorySessionStore()
assertEquals(rawValueToSessionHeadersValue(), converter);
Review Comment:
Pre-existing, but since this PR already reworks most of this file: the test
name says `Identity` while the assertion is `rawValueToSessionHeadersValue()`.
Maybe rename to something like
`shouldReturnSessionHeadersConverterForPlainToHeadersInMemorySessionStore`
while we're here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
