aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3669066050
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java:
##########
@@ -67,22 +67,22 @@ public class SessionToHeadersStoreAdapter implements
SessionStore<Bytes, byte[]>
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes
key,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- return new SessionToHeadersIteratorAdapter(
+ return MappingKeyValueIteratorAdapter.sessionToHeaders(
store.findSessions(key, earliestSessionEndTime,
latestSessionStartTime));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final long
earliestSessionEndTime,
final long
latestSessionEndTime) {
- return new SessionToHeadersIteratorAdapter(
+ return MappingKeyValueIteratorAdapter.sessionToHeaders(
Review Comment:
This `findSessions(long, long)` overload is now the only session method with
no conversion test — all the others and the WindowRangeQuery branch got one.
Please add the same `assertAddsEmptyHeaders` test for it.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapterTest.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * 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.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.kstream.internals.SessionWindow;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import static
org.apache.kafka.streams.state.HeadersBytesStore.convertFromPlainToHeaderFormat;
+import static
org.apache.kafka.streams.state.HeadersBytesStore.convertToHeaderFormat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
+public class MappingKeyValueIteratorAdapterTest {
+
+ private static final Bytes KEY = Bytes.wrap("key".getBytes());
+ private static final byte[] RAW_VALUE = "value".getBytes();
+ private static final long TIMESTAMP = 42L;
+ private static final Windowed<Bytes> SESSION_KEY =
+ new Windowed<>(KEY, new SessionWindow(10L, 20L));
+
+ @Mock
+ private KeyValueIterator<Bytes, byte[]> inner;
+
+ @Mock
+ private KeyValueIterator<Windowed<Bytes>, byte[]> sessionInner;
+
+ @Mock
+ private KeyValueIterator<Long, byte[]> windowInner;
+
+ @Test
+ public void plainToHeadersShouldConvertValueOnNext() {
+ when(inner.hasNext()).thenReturn(true);
+ when(inner.next()).thenReturn(KeyValue.pair(KEY, RAW_VALUE));
+
+ final KeyValueIterator<Bytes, byte[]> adapter =
+ MappingKeyValueIteratorAdapter.plainToHeaders(inner);
+
+ assertTrue(adapter.hasNext());
+ final KeyValue<Bytes, byte[]> result = adapter.next();
+ assertEquals(KEY, result.key);
+ assertArrayEquals(convertFromPlainToHeaderFormat(RAW_VALUE),
result.value);
+ }
+
+ @Test
+ public void timestampedToHeadersShouldConvertValueOnNext() {
+ when(inner.hasNext()).thenReturn(true);
+ when(inner.next()).thenReturn(KeyValue.pair(KEY, RAW_VALUE));
+
+ final KeyValueIterator<Bytes, byte[]> adapter =
+ MappingKeyValueIteratorAdapter.timestampedToHeaders(inner);
+
+ assertTrue(adapter.hasNext());
+ final KeyValue<Bytes, byte[]> result = adapter.next();
+ assertEquals(KEY, result.key);
+ assertArrayEquals(convertToHeaderFormat(RAW_VALUE), result.value);
+ }
+
+ @Test
+ public void sessionToHeadersShouldConvertValueOnNext() {
+ when(sessionInner.hasNext()).thenReturn(true);
+ when(sessionInner.next()).thenReturn(KeyValue.pair(SESSION_KEY,
RAW_VALUE));
+
+ final KeyValueIterator<Windowed<Bytes>, byte[]> adapter =
+ MappingKeyValueIteratorAdapter.sessionToHeaders(sessionInner);
+
+ assertTrue(adapter.hasNext());
+ final KeyValue<Windowed<Bytes>, byte[]> result = adapter.next();
+ assertEquals(SESSION_KEY, result.key);
+ assertArrayEquals(convertToHeaderFormat(RAW_VALUE), result.value);
+ }
+
+ @Test
+ public void plainToHeadersWindowShouldConvertValueOnNext() {
+ when(windowInner.hasNext()).thenReturn(true);
+ when(windowInner.next()).thenReturn(KeyValue.pair(TIMESTAMP,
RAW_VALUE));
+
+ final WindowStoreIterator<byte[]> adapter =
+ MappingKeyValueIteratorAdapter.plainToHeadersWindow(windowInner);
+
+ assertInstanceOf(WindowStoreIterator.class, adapter);
Review Comment:
The factory's return type is already `WindowStoreIterator<byte[]>`, so this
can only fail if it returned null. Drop it (and the same line 128) — the array
assertion below is the one that matters.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java:
##########
@@ -100,6 +102,197 @@ public void tearDown() {
}
}
+ @Test
+ public void shouldConvertValueOnFetch() {
Review Comment:
These 14 `shouldConvertValueOn*` tests are the same body with a different
iterator call. One `@ParameterizedTest` over the accessors would keep the
coverage and let both `assertConvertedValue` overloads go.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java:
##########
@@ -101,6 +103,197 @@ public void tearDown() {
}
}
+ @Test
+ public void shouldConvertValueOnFetch() {
Review Comment:
Same as in `PlainToHeadersWindowStoreAdapterTest`: these 14 tests differ
only in the iterator call, so one `@ParameterizedTest` over the accessors would
replace them.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java:
##########
@@ -0,0 +1,420 @@
+/*
+ * 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.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+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.streams.query.RangeQuery;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.TimestampedBytesStore;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static
org.apache.kafka.streams.state.HeadersBytesStore.convertToHeaderFormat;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.STRICT_STUBS)
+public class TimestampedToHeadersStoreAdapterTest {
+
+ @Mock(extraInterfaces = TimestampedBytesStore.class)
+ private KeyValueStore<Bytes, byte[]> mockStore;
+
+ @Mock
+ private KeyValueIterator<Bytes, byte[]> mockIterator;
+
+ private TimestampedToHeadersStoreAdapter adapter;
+
+ private TimestampedToHeadersStoreAdapter createAdapter() {
+ when(mockStore.persistent()).thenReturn(true);
+ return new TimestampedToHeadersStoreAdapter(mockStore);
+ }
+
+ private void assertConvertsTimestampedToHeaders(final
KeyValueIterator<Bytes, byte[]> result) {
+ final Bytes key = new Bytes("k".getBytes());
+ final byte[] timestampedValue = "value".getBytes();
+ when(mockIterator.hasNext()).thenReturn(true);
+ when(mockIterator.next()).thenReturn(KeyValue.pair(key,
timestampedValue));
+
+ assertTrue(result.hasNext());
+ final KeyValue<Bytes, byte[]> entry = result.next();
+ assertEquals(key, entry.key);
+ // Timestamped format only prepends empty headers; the plain
conversion would also insert
+ // an 8-byte timestamp, so this array comparison proves
timestampedToHeaders was wired.
+ assertArrayEquals(convertToHeaderFormat(timestampedValue),
entry.value);
+ }
+
+ @Test
+ public void shouldThrowIfStoreIsNotPersistent() {
+ when(mockStore.persistent()).thenReturn(false);
+
+ final IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new TimestampedToHeadersStoreAdapter(mockStore)
+ );
+
+ assertTrue(exception.getMessage().contains("Provided store must be a
persistent store"));
+ }
+
+ @Test
+ public void shouldThrowIfStoreIsNotTimestamped() {
+ @SuppressWarnings("unchecked")
+ final KeyValueStore<Bytes, byte[]> plainStore =
mock(KeyValueStore.class);
+ when(plainStore.persistent()).thenReturn(true);
+
+ final IllegalArgumentException exception = assertThrows(
+ IllegalArgumentException.class,
+ () -> new TimestampedToHeadersStoreAdapter(plainStore)
+ );
+
+ assertTrue(exception.getMessage().contains("Provided store must be a
timestamped store"));
+ }
+
+ @Test
+ public void shouldPutRawTimestampedValueToStore() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] timestampedValue = "value".getBytes();
+ final byte[] valueWithHeaders =
convertToHeaderFormat(timestampedValue);
+
+ adapter.put(key, valueWithHeaders);
+
+ verify(mockStore).put(eq(key), eq(timestampedValue));
+ }
+
+ @Test
+ public void shouldGetAndConvertToHeaderFormat() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] timestampedValue = "value".getBytes();
+ when(mockStore.get(key)).thenReturn(timestampedValue);
+
+ final byte[] result = adapter.get(key);
+
+ assertArrayEquals(convertToHeaderFormat(timestampedValue), result);
+ }
+
+ @Test
+ public void shouldReturnNullWhenStoreReturnsNull() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("key".getBytes());
+ when(mockStore.get(key)).thenReturn(null);
+
+ final byte[] result = adapter.get(key);
+
+ assertNull(result);
+ }
+
+ @Test
+ public void shouldPutIfAbsentAndConvertResult() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] timestampedValue = "value".getBytes();
+ final byte[] valueWithHeaders =
convertToHeaderFormat(timestampedValue);
+ final byte[] oldTimestampedValue = "oldValue".getBytes();
+ when(mockStore.putIfAbsent(eq(key),
eq(timestampedValue))).thenReturn(oldTimestampedValue);
+
+ final byte[] result = adapter.putIfAbsent(key, valueWithHeaders);
+
+ assertArrayEquals(convertToHeaderFormat(oldTimestampedValue), result);
+ }
+
+ @Test
+ public void shouldDeleteAndConvertResult() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("key".getBytes());
+ final byte[] oldTimestampedValue = "oldValue".getBytes();
+ when(mockStore.delete(key)).thenReturn(oldTimestampedValue);
+
+ final byte[] result = adapter.delete(key);
+
+ assertArrayEquals(convertToHeaderFormat(oldTimestampedValue), result);
+ }
+
+ @Test
+ public void shouldPutAllEntries() {
+ adapter = createAdapter();
+ final Bytes key1 = new Bytes("key1".getBytes());
+ final Bytes key2 = new Bytes("key2".getBytes());
+ final byte[] value1 = convertToHeaderFormat("value1".getBytes());
+ final byte[] value2 = convertToHeaderFormat("value2".getBytes());
+
+ final List<KeyValue<Bytes, byte[]>> entries = Arrays.asList(
+ KeyValue.pair(key1, value1),
+ KeyValue.pair(key2, value2)
+ );
+
+ adapter.putAll(entries);
+
+ verify(mockStore).put(eq(key1), eq("value1".getBytes()));
+ verify(mockStore).put(eq(key2), eq("value2".getBytes()));
+ }
+
+ @Test
+ public void shouldWrapRangeIterator() {
+ adapter = createAdapter();
+ final Bytes from = new Bytes("a".getBytes());
+ final Bytes to = new Bytes("z".getBytes());
+ when(mockStore.range(from, to)).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.range(from, to);
+
+ assertNotNull(result);
+ assertConvertsTimestampedToHeaders(result);
+ }
+
+ @Test
+ public void shouldWrapReverseRangeIterator() {
+ adapter = createAdapter();
+ final Bytes from = new Bytes("a".getBytes());
+ final Bytes to = new Bytes("z".getBytes());
+ when(mockStore.reverseRange(from, to)).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result =
adapter.reverseRange(from, to);
+
+ assertNotNull(result);
+ assertConvertsTimestampedToHeaders(result);
+ }
+
+ @Test
+ public void shouldWrapAllIterator() {
+ adapter = createAdapter();
+ when(mockStore.all()).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.all();
+
+ assertNotNull(result);
+ assertConvertsTimestampedToHeaders(result);
+ }
+
+ @Test
+ public void shouldWrapReverseAllIterator() {
+ adapter = createAdapter();
+ when(mockStore.reverseAll()).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result = adapter.reverseAll();
+
+ assertNotNull(result);
+ assertConvertsTimestampedToHeaders(result);
+ }
+
+ @Test
+ public void shouldWrapPrefixScanIterator() {
+ adapter = createAdapter();
+ when(mockStore.prefixScan(any(), any())).thenReturn(mockIterator);
+
+ final KeyValueIterator<Bytes, byte[]> result =
adapter.prefixScan("prefix", (topic, data) -> data.getBytes());
+
+ assertNotNull(result);
+ assertConvertsTimestampedToHeaders(result);
+ }
+
+ @Test
+ public void shouldHandleKeyQuery() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] timestampedValue = "test-value".getBytes();
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forResult(timestampedValue);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result = adapter.query(query,
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ assertArrayEquals(convertToHeaderFormat(timestampedValue),
result.getResult());
+ }
+
+ @Test
+ public void shouldHandleKeyQueryWithNullResult() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("test-key".getBytes());
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult = QueryResult.forResult(null);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result = adapter.query(query,
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertTrue(result.isSuccess());
+ assertNull(result.getResult());
+ }
+
+ @Test
+ public void shouldHandleFailedKeyQuery() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("test-key".getBytes());
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forUnknownQueryType(query, mockStore);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result = adapter.query(query,
PositionBound.unbounded(), new QueryConfig(false));
+
+ assertFalse(result.isSuccess());
+ }
+
+ @Test
+ public void shouldHandleRangeQuery() {
+ adapter = createAdapter();
+ final RangeQuery<Bytes, byte[]> query = RangeQuery.withRange(
+ new Bytes("a".getBytes()),
+ new Bytes("z".getBytes())
+ );
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> mockResult =
QueryResult.forResult(mockIterator);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> result =
adapter.query(
+ query,
+ PositionBound.unbounded(),
+ new QueryConfig(false)
+ );
+
+ assertTrue(result.isSuccess());
+ assertNotNull(result.getResult());
+ assertConvertsTimestampedToHeaders(result.getResult());
+ }
+
+ @Test
+ public void shouldCollectExecutionInfoForKeyQuery() {
+ adapter = createAdapter();
+ final Bytes key = new Bytes("test-key".getBytes());
+ final byte[] timestampedValue = "test-value".getBytes();
+ final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
+
+ final QueryResult<byte[]> mockResult =
QueryResult.forResult(timestampedValue);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<byte[]> result = adapter.query(query,
PositionBound.unbounded(), new QueryConfig(true));
+
+ assertTrue(result.isSuccess());
+ 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");
+
assertTrue(executionInfo.contains(TimestampedToHeadersStoreAdapter.class.getName()),
+ "Expected execution info to mention
TimestampedToHeadersStoreAdapter");
+ }
+
+ @Test
+ public void shouldCollectExecutionInfoForRangeQuery() {
+ adapter = createAdapter();
+ final RangeQuery<Bytes, byte[]> query = RangeQuery.withRange(
+ new Bytes("a".getBytes()),
+ new Bytes("z".getBytes())
+ );
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> mockResult =
QueryResult.forResult(mockIterator);
+ when(mockStore.query(eq(query), any(PositionBound.class),
any(QueryConfig.class)))
+ .thenReturn(mockResult);
+
+ final QueryResult<KeyValueIterator<Bytes, byte[]>> result =
adapter.query(
+ query,
+ PositionBound.unbounded(),
+ new QueryConfig(true)
+ );
+
+ assertTrue(result.isSuccess());
+ 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");
+ }
+
+ @Test
+ public void shouldDelegateOtherQueryTypesToUnderlyingStore() {
Review Comment:
This is the same test as `shouldHandleFailedKeyQuery` (line 284): same query
type, same stub, same assertion, only the key string differs. Drop one of the
two.
--
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]
