You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nileshkumar3 (via GitHub)" <gi...@apache.org> on 2026/03/20 02:18:32 UTC
[PR] KAFKA-20224: Deduplicate ToHeaders iterator adapters with MappingKeyValueIteratorAdapter [kafka]
nileshkumar3 opened a new pull request, #21830:
URL: https://github.com/apache/kafka/pull/21830
We had three almost-identical iterators that only wrapped the inner iterator and ran a different HeadersBytesStore conversion on the value bytes. I pulled that into a small MappingKeyValueIteratorAdapter that takes a Function<byte[], byte[]> and left the three existing adapter classes as thin subclasses so names and call sites stay the same. Behavior should be unchanged; the usual ToHeaders iterator/store/window tests still pass.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-5860468363
@aliehsaeedii gentle ping to review.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4934288764
One more thing I noticed now: the same duplication you just removed still lives in the two *window-store* iterator adapters:
- `PlainToHeadersWindowStoreIteratorAdapter` (top-level class)
- `TimestampedWindowToHeadersWindowStoreIteratorAdapter` (private nested class in `TimestampedToHeadersWindowStoreAdapter`)
Both have no body beyond `extends MappingKeyValueIteratorAdapter<Long> implements WindowStoreIterator<byte[]>` and differ only by the mapper (`convertFromPlainToHeaderFormat` vs `convertToHeaderFormat`) — the only reason they exist as subclasses is to carry the `WindowStoreIterator` marker on the return type. Since this is exactly the pattern this PR set out to eliminate, I'd suggest finishing the job here for consistency: add window-flavored factories to `MappingKeyValueIteratorAdapter` backed by a single private inner subclass, e.g.
```java
private static final class WindowStoreIteratorAdapter
extends MappingKeyValueIteratorAdapter<Long>
implements WindowStoreIterator<byte[]> {
WindowStoreIteratorAdapter(final KeyValueIterator<Long, byte[]> inner, final Function<byte[], byte[]> mapper) {
super(inner, mapper);
}
}
static WindowStoreIterator<byte[]> plainToHeadersWindow(final KeyValueIterator<Long, byte[]> inner) {
return new WindowStoreIteratorAdapter(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
}
static WindowStoreIterator<byte[]> timestampedToHeadersWindow(final KeyValueIterator<Long, byte[]> inner) {
return new WindowStoreIteratorAdapter(inner, HeadersBytesStore::convertToHeaderFormat);
}
```
Then both window iterator classes can be deleted and their call sites updated. No behavior change.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4976547473
Thanks @aliehsaeedii , I added TimestampedToHeadersStoreAdapterTest and value-conversion coverage for fetch, backwardFetch, and fetchAll in both window-store adapter tests. please have another look.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4412531614
@mjsax could you please review this once you get a chance.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3557834704
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapterTest.java:
##########
@@ -180,7 +180,7 @@ public void shouldWrapRangeIterator() {
final KeyValueIterator<Bytes, byte[]> result = adapter.range(from, to);
assertNotNull(result);
- assertTrue(result instanceof PlainToHeadersIteratorAdapter);
+ assertTrue(result instanceof MappingKeyValueIteratorAdapter);
Review Comment:
This `instanceof MappingKeyValueIteratorAdapter` check no longer proves the right conversion was wired — all three factories return this same type, so if `range()` accidentally called `timestampedToHeaders(...)` this would still pass. Consider asserting on the converted bytes (call `next()` and check the header prefix). Same for the other instanceof checks in this file.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java:
##########
@@ -110,7 +110,7 @@ public void shouldWrapFindSessionsIterator() {
final KeyValueIterator<Windowed<Bytes>, byte[]> innerIter = mock(KeyValueIterator.class);
when(innerStore.findSessions(KEY, 10L, 20L)).thenReturn(innerIter);
final KeyValueIterator<Windowed<Bytes>, byte[]> result = adapter.findSessions(KEY, 10L, 20L);
- assertInstanceOf(SessionToHeadersIteratorAdapter.class, result);
+ assertInstanceOf(MappingKeyValueIteratorAdapter.class, result);
Review Comment:
Same point as in PlainToHeadersStoreAdapterTest: `assertInstanceOf(MappingKeyValueIteratorAdapter.class, ...)` no longer distinguishes the session mapping from the plain/timestamped ones. Assert on the converted value to keep the check meaningful.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.SessionStoreWithHeaders;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and {@link TimestampedKeyValueStore}.
Review Comment:
`sessionToHeaders` has a direct next()/null-handling test but `plainToHeaders` and `timestampedToHeaders` don't. Since it's one class now, a small MappingKeyValueIteratorAdapterTest covering all three factories would close that gap.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4947080407
Thanks @aliehsaeedii for the review. Fixed all the review comments, please have another look.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-6014665067
@frankvicky could we merge this PR? thanks.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4145952664
@mjsax just a gentle follow-up on this PR. I’ve addressed the previous feedback and cleaned up the implementation (including removing the subclasses). Would appreciate a review when you have time. Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3564456441
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersStoreAdapterTest.java:
##########
@@ -180,7 +180,7 @@ public void shouldWrapRangeIterator() {
final KeyValueIterator<Bytes, byte[]> result = adapter.range(from, to);
assertNotNull(result);
- assertTrue(result instanceof PlainToHeadersIteratorAdapter);
+ assertTrue(result instanceof MappingKeyValueIteratorAdapter);
Review Comment:
switched it to assert on the converted bytes.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3564458680
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.SessionStoreWithHeaders;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and {@link TimestampedKeyValueStore}.
Review Comment:
Makes sense, added a MappingKeyValueIteratorAdapterTest that covers all three factories
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-5553846873
@aliehsaeedii Sorry for the delay. I addressed all review comments. Can you please review once you get a chance.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4550631236
@mjsax gentle ping to review this one.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4659528929
@mjsax @frankvicky gentle ping on this.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3575931654
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapterTest.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.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.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 Windowed<Bytes> SESSION_KEY =
+ new Windowed<>(KEY, new SessionWindow(10L, 20L));
+
+ @Mock
+ private KeyValueIterator<Bytes, byte[]> inner;
+
+ @Mock
+ private KeyValueIterator<Windowed<Bytes>, byte[]> sessionInner;
+
+ @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() {
Review Comment:
Good catch — added plainToHeadersWindowShouldConvertValueOnNext and timestampedToHeadersWindowShouldConvertValueOnNext tests.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r2963824119
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersIteratorAdapter.java:
##########
@@ -16,47 +16,19 @@
*/
package org.apache.kafka.streams.state.internals;
-import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.HeadersBytesStore;
import org.apache.kafka.streams.state.KeyValueIterator;
-import static org.apache.kafka.streams.state.HeadersBytesStore.convertFromPlainToHeaderFormat;
-
/**
* This class is used to ensure backward compatibility at DSL level between
* {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders} and
* plain {@link org.apache.kafka.streams.state.KeyValueStore}.
*
* @see PlainToHeadersStoreAdapter
*/
-
-class PlainToHeadersIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
- private final KeyValueIterator<K, byte[]> innerIterator;
+class PlainToHeadersIteratorAdapter<K> extends MappingKeyValueIteratorAdapter<K> {
Review Comment:
Do we even need to keep these sub-classes? Or could the calling code just do `newMappingKeyValueIteratorAdapter(...)` instead?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4097771397
> Thanks for the PR only some nits.
Thanks for reviewing, replaced with import.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4479622841
@frankvicky All the previous comments have been incorporated. please have a look when you get a chance.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r2965192102
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
Review Comment:
nit: replace full qualified with import
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link org.apache.kafka.streams.state.KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
+ * and {@link org.apache.kafka.streams.state.TimestampedKeyValueStore}.
Review Comment:
nit: replace full qualified with import
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link org.apache.kafka.streams.state.KeyValueStore}: values are wrapped with empty headers
Review Comment:
nit: replace full qualified with import
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link org.apache.kafka.streams.state.KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders}
+ * and {@link org.apache.kafka.streams.state.TimestampedKeyValueStore}.
+ *
+ * @see TimestampedToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> timestampedToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link org.apache.kafka.streams.state.SessionStoreWithHeaders}
+ * and {@link org.apache.kafka.streams.state.SessionStore}.
Review Comment:
nit: replace full qualified with import
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r2964150029
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersIteratorAdapter.java:
##########
@@ -16,47 +16,19 @@
*/
package org.apache.kafka.streams.state.internals;
-import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.HeadersBytesStore;
import org.apache.kafka.streams.state.KeyValueIterator;
-import static org.apache.kafka.streams.state.HeadersBytesStore.convertFromPlainToHeaderFormat;
-
/**
* This class is used to ensure backward compatibility at DSL level between
* {@link org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders} and
* plain {@link org.apache.kafka.streams.state.KeyValueStore}.
*
* @see PlainToHeadersStoreAdapter
*/
-
-class PlainToHeadersIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
- private final KeyValueIterator<K, byte[]> innerIterator;
+class PlainToHeadersIteratorAdapter<K> extends MappingKeyValueIteratorAdapter<K> {
Review Comment:
Good point. These subclasses no longer add any behavior beyond selecting the mapping function, so they could be removed and the call could instantiate MappingKeyValueIteratorAdapter directly. I kept them earlier here mainly to limit churn, but I cleaned this.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3578459239
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapter.java:
##########
@@ -201,29 +201,29 @@ public byte[] get(final Bytes key) {
@Override
public KeyValueIterator<Bytes, byte[]> range(final Bytes from,
final Bytes to) {
- return new TimestampedToHeadersIteratorAdapter<>(store.range(from, to));
+ return MappingKeyValueIteratorAdapter.timestampedToHeaders(store.range(from, to));
Review Comment:
There is no `TimestampedToHeadersStoreAdapterTest,` so no test asserts these methods wire the timestampedToHeaders conversion rather than plainToHeaders.
The plain and session siblings just got value-conversion checks, could we add the analogous value-conversion test for the timestamped adapter.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java:
##########
@@ -80,76 +80,76 @@ public byte[] fetch(final Bytes key, final long timestamp) {
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom, final long timeTo) {
- return new PlainToHeadersWindowStoreIteratorAdapter(store.fetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeadersWindow(store.fetch(key, timeFrom, timeTo));
Review Comment:
The window-store adapter tests only exercise `query()`, never `fetch`/`backwardFetch`/`fetchAll`, so the methods now calling the new window factories aren't covered. A test that calls `fetch` and asserts the converted value would close the gap.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3572540331
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapterTest.java:
##########
@@ -0,0 +1,149 @@
+/*
+ * 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.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.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 Windowed<Bytes> SESSION_KEY =
+ new Windowed<>(KEY, new SessionWindow(10L, 20L));
+
+ @Mock
+ private KeyValueIterator<Bytes, byte[]> inner;
+
+ @Mock
+ private KeyValueIterator<Windowed<Bytes>, byte[]> sessionInner;
+
+ @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() {
Review Comment:
This covers plainToHeaders, timestampedToHeaders and sessionToHeaders, but not the two window factories plainToHeadersWindow and timestampedToHeadersWindow. Add a test for those too, asserting the value conversion and that the result is a WindowStoreIterator.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3712783504
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.SessionStoreWithHeaders;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertFromPlainToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and {@link TimestampedKeyValueStore}.
+ *
+ * @see TimestampedToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> timestampedToHeaders(final KeyValueIterator<K, byte[]> inner) {
+ return new MappingKeyValueIteratorAdapter<>(inner, HeadersBytesStore::convertToHeaderFormat);
+ }
+
+ /**
+ * Ensures backward compatibility between {@link SessionStoreWithHeaders}
+ * and {@link SessionStore}.
+ *
+ * @see SessionToHeadersStoreAdapter
+ */
+ static KeyValueIterator<Windowed<Bytes>, byte[]> sessionToHeaders(
Review Comment:
This uses the same mapper as `timestampedToHeaders` above, so it's just a type-narrowed alias. Can the session call sites call `timestampedToHeaders` instead?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/MappingKeyValueIteratorAdapter.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.state.HeadersBytesStore;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.SessionStoreWithHeaders;
+import org.apache.kafka.streams.state.TimestampedKeyValueStore;
+import org.apache.kafka.streams.state.TimestampedKeyValueStoreWithHeaders;
+import org.apache.kafka.streams.state.WindowStoreIterator;
+
+import java.util.function.Function;
+
+/**
+ * Delegates to an inner {@link KeyValueIterator} and maps each value byte array
+ * through the given function (e.g. header-format conversion).
+ */
+class MappingKeyValueIteratorAdapter<K> implements KeyValueIterator<K, byte[]> {
+
+ private final KeyValueIterator<K, byte[]> innerIterator;
+ private final Function<byte[], byte[]> valueMapper;
+
+ MappingKeyValueIteratorAdapter(
+ final KeyValueIterator<K, byte[]> innerIterator,
+ final Function<byte[], byte[]> valueMapper
+ ) {
+ this.innerIterator = innerIterator;
+ this.valueMapper = valueMapper;
+ }
+
+ /**
+ * Ensures backward compatibility between {@link TimestampedKeyValueStoreWithHeaders}
+ * and plain {@link KeyValueStore}: values are wrapped with empty headers
+ * and timestamp {@code -1}.
+ *
+ * @see PlainToHeadersStoreAdapter
+ */
+ static <K> KeyValueIterator<K, byte[]> plainToHeaders(final KeyValueIterator<K, byte[]> inner) {
Review Comment:
The doc only mentions the key-value adapter, but `plainToHeaders` is also used by `PlainToHeadersWindowStoreAdapter` for its key-range, `fetchAll` and `all()` iterators. Should we add that `@see` (same for `timestampedToHeaders`).
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java:
##########
@@ -333,54 +341,83 @@ public void shouldDelegateOtherQueryTypesToStore() {
@Test
public void shouldCollectExecutionInfoForKeyQuery() {
+ adapter = createAdapter();
final Bytes key = new Bytes("test-key".getBytes());
- final byte[] rawTimestampedValue =
- new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
+ final byte[] timestampedValue = "test-value".getBytes();
final KeyQuery<Bytes, byte[]> query = KeyQuery.withKey(key);
- final QueryResult<byte[]> mockResult = QueryResult.forResult(rawTimestampedValue);
+ 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));
+ 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");
+ assertFalse(result.getExecutionInfo().isEmpty(), "Expected execution info to be collected");
final String executionInfo = String.join("\n", result.getExecutionInfo());
- assertTrue(executionInfo.contains("Handled in"));
- assertTrue(executionInfo.contains(TimestampedToHeadersStoreAdapter.class.getName()));
+ 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() {
Review Comment:
`addExecutionInfo` sits outside the query-type branches, so this hits the same line as `shouldCollectExecutionInfoForKeyQuery`, and the RangeQuery branch is already covered by `shouldHandleRangeQuery`. Drop it, or at least add the `contains(className)` check the KeyQuery test has.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersStoreAdapterTest.java:
##########
@@ -37,268 +36,276 @@
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.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
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.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import static org.mockito.Mockito.withSettings;
@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;
- @SuppressWarnings("unchecked")
- private KeyValueStore<Bytes, byte[]> mockStore;
-
private TimestampedToHeadersStoreAdapter adapter;
- @SuppressWarnings("unchecked")
- @BeforeEach
- public void setUp() {
- mockStore = mock(KeyValueStore.class, withSettings().extraInterfaces(TimestampedBytesStore.class));
- // lenient: this fixture stub is consumed by the adapter constructor for most tests, but the
- // constructor-validation tests build their own store and never touch this one.
- lenient().when(mockStore.persistent()).thenReturn(true);
- adapter = new TimestampedToHeadersStoreAdapter(mockStore);
+ 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
- @SuppressWarnings("unchecked")
public void shouldThrowIfStoreIsNotPersistent() {
- final KeyValueStore<Bytes, byte[]> nonPersistentStore =
- mock(KeyValueStore.class, withSettings().extraInterfaces(TimestampedBytesStore.class));
- when(nonPersistentStore.persistent()).thenReturn(false);
+ when(mockStore.persistent()).thenReturn(false);
final IllegalArgumentException exception = assertThrows(
IllegalArgumentException.class,
- () -> new TimestampedToHeadersStoreAdapter(nonPersistentStore)
+ () -> new TimestampedToHeadersStoreAdapter(mockStore)
);
assertTrue(exception.getMessage().contains("Provided store must be a persistent store"));
}
@Test
- @SuppressWarnings("unchecked")
public void shouldThrowIfStoreIsNotTimestamped() {
- final KeyValueStore<Bytes, byte[]> nonTimestampedStore = mock(KeyValueStore.class);
- when(nonTimestampedStore.persistent()).thenReturn(true);
+ @SuppressWarnings("unchecked")
+ final KeyValueStore<Bytes, byte[]> plainStore = mock(KeyValueStore.class);
+ when(plainStore.persistent()).thenReturn(true);
final IllegalArgumentException exception = assertThrows(
IllegalArgumentException.class,
- () -> new TimestampedToHeadersStoreAdapter(nonTimestampedStore)
+ () -> 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[] rawTimestampedValue =
- new byte[] {0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'};
- final byte[] valueWithHeaders = convertToHeaderFormat(rawTimestampedValue);
+ final byte[] timestampedValue = "value".getBytes();
Review Comment:
This was `{0, 0, 0, 0, 0, 0, 0, 42, 'v', 'a', 'l'}` before. `"value".getBytes()` has no 8-byte timestamp prefix, so the test no longer shows the format its name refers to — please keep the old fixture here and in the other put/get/delete tests.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky merged PR #21830:
URL: https://github.com/apache/kafka/pull/21830
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4115257351
@mjsax @frankvicky thanks for the review, fixed all the comments. Let me know if anything else need to be changed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4132912055
> Thanks @nileshkumar3 for the PR. LGTM.
Thanks @aliehsaeedii for reviewing.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-4193601726
@mjsax No urgency — when it fits your queue, could you take another look at the PR ,I’ve incorporated the earlier review notes. Thank you.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3557995080
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreIteratorAdapter.java:
##########
@@ -16,20 +16,21 @@
*/
Review Comment:
Do we still neeed this class?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreIteratorAdapter.java:
##########
@@ -16,20 +16,21 @@
*/
Review Comment:
Do we still neeed this class?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3564458111
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapterTest.java:
##########
@@ -110,7 +110,7 @@ public void shouldWrapFindSessionsIterator() {
final KeyValueIterator<Windowed<Bytes>, byte[]> innerIter = mock(KeyValueIterator.class);
when(innerStore.findSessions(KEY, 10L, 20L)).thenReturn(innerIter);
final KeyValueIterator<Windowed<Bytes>, byte[]> result = adapter.findSessions(KEY, 10L, 20L);
- assertInstanceOf(SessionToHeadersIteratorAdapter.class, result);
+ assertInstanceOf(MappingKeyValueIteratorAdapter.class, result);
Review Comment:
yes, fixed same 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-5080892172
@aliehsaeedii, addressed all three — added value-conversion tests for the WindowRangeQuery branch and the remaining fetch/backwardFetch/all overloads in both window adapters. please take a look when you get a chance.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on PR #21830:
URL: https://github.com/apache/kafka/pull/21830#issuecomment-5125908082
Thanks @aliehsaeedii ! Addressed all comments:
Collapsed the 14 shouldConvertValueOn* tests into one @ParameterizedTest in both window-adapter tests (dropped both assertConvertedValue overloads).
Removed the redundant assertInstanceOf(WindowStoreIterator...) checks in MappingKeyValueIteratorAdapterTest.
Dropped the duplicate shouldDelegateOtherQueryTypesToUnderlyingStore.
Added a conversion test for the findSessions(long, long) overload.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20224: Refactor Streams iterator adapters to reduce duplication [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21830:
URL: https://github.com/apache/kafka/pull/21830#discussion_r3645523879
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapter.java:
##########
@@ -81,76 +81,76 @@ public byte[] fetch(final Bytes key, final long timestamp) {
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom, final long timeTo) {
- return new TimestampedWindowToHeadersWindowStoreIteratorAdapter(store.fetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.timestampedToHeadersWindow(store.fetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {
- return new TimestampedWindowToHeadersWindowStoreIteratorAdapter(store.fetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.timestampedToHeadersWindow(store.fetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> backwardFetch(final Bytes key, final long timeFrom, final long timeTo) {
- return new TimestampedWindowToHeadersWindowStoreIteratorAdapter(store.backwardFetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.timestampedToHeadersWindow(store.backwardFetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> backwardFetch(final Bytes key, final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {
- return new TimestampedWindowToHeadersWindowStoreIteratorAdapter(store.backwardFetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.timestampedToHeadersWindow(store.backwardFetch(key, timeFrom, timeTo));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom, final Bytes keyTo,
final long timeFrom, final long timeTo) {
- return new TimestampedToHeadersIteratorAdapter<>(store.fetch(keyFrom, keyTo, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.timestampedToHeaders(store.fetch(keyFrom, keyTo, timeFrom, timeTo));
Review Comment:
Same gap as the plain window adapter: only the `long`-bound fetch/backwardFetch/fetchAll are tested. The Instant overloads, key-range fetch/backwardFetch, backwardFetchAll, all(), and backwardAll() have no conversion test. Add them so a wrong mapper here would fail.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/SessionToHeadersStoreAdapter.java:
##########
@@ -197,7 +197,7 @@ public <R> QueryResult<R> query(final Query<R> query,
final WindowRangeQuery<Bytes, byte[]> windowRangeQuery = (WindowRangeQuery<Bytes, byte[]>) query;
final QueryResult<KeyValueIterator<Windowed<Bytes>, byte[]>> rawResult = store.query(windowRangeQuery, positionBound, config);
if (rawResult.isSuccess()) {
- final KeyValueIterator<Windowed<Bytes>, byte[]> wrappedIterator = new SessionToHeadersIteratorAdapter(rawResult.getResult());
+ final KeyValueIterator<Windowed<Bytes>, byte[]> wrappedIterator = MappingKeyValueIteratorAdapter.sessionToHeaders(rawResult.getResult());
Review Comment:
This WindowRangeQuery branch has no test, so nothing checks that `sessionToHeaders` is wired here (or that it converts at all). Add a query test with a WindowRangeQuery that asserts the converted value.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapter.java:
##########
@@ -80,76 +80,76 @@ public byte[] fetch(final Bytes key, final long timestamp) {
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom, final long timeTo) {
- return new PlainToHeadersWindowStoreIteratorAdapter(store.fetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeadersWindow(store.fetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {
- return new PlainToHeadersWindowStoreIteratorAdapter(store.fetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeadersWindow(store.fetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> backwardFetch(final Bytes key, final long timeFrom, final long timeTo) {
- return new PlainToHeadersWindowStoreIteratorAdapter(store.backwardFetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeadersWindow(store.backwardFetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> backwardFetch(final Bytes key, final Instant timeFrom, final Instant timeTo) throws IllegalArgumentException {
- return new PlainToHeadersWindowStoreIteratorAdapter(store.backwardFetch(key, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeadersWindow(store.backwardFetch(key, timeFrom, timeTo));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom, final Bytes keyTo,
final long timeFrom, final long timeTo) {
- return new PlainToHeadersIteratorAdapter<>(store.fetch(keyFrom, keyTo, timeFrom, timeTo));
+ return MappingKeyValueIteratorAdapter.plainToHeaders(store.fetch(keyFrom, keyTo, timeFrom, timeTo));
Review Comment:
Only fetch/backwardFetch/fetchAll with `long` bounds get a conversion test. The Instant overloads, the key-range fetch/backwardFetch, backwardFetchAll, all(), and backwardAll() aren't exercised with a populated store, so a wrong mapper on any of these lines wouldn't be caught. Add conversion tests for these too.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org