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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
