aliehsaeedii commented on code in PR #22156:
URL: https://github.com/apache/kafka/pull/22156#discussion_r3472426526
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -248,38 +253,40 @@ private void emitNonJoinedOuterRecords(final
KeyValueStore<TimestampedKeyAndJoin
continue;
}
- final LeftOrRightValue<VLeft, VRight> leftOrRightValue =
nextKeyValue.value;
- forwardNonJoinedOuterRecords(record,
timestampedKeyAndJoinSide, leftOrRightValue);
+ forwardNonJoinedOuterRecords(record,
timestampedKeyAndJoinSide, nextKeyValue.value);
if (prevKey != null &&
!prevKey.equals(timestampedKeyAndJoinSide)) {
// blind-delete the previous key from the outer window
store now it is emitted;
// we do this because this delete would remove the
whole list of values of the same key,
// and hence if we delete eagerly and then fail, we
would miss emitting join results of the later
// values in the list.
// we do not use delete() calls since it would incur
extra get()
- store.put(prevKey, null);
+ outerJoinStoreWrapper.put(prevKey, null, null, 0L);
Review Comment:
With the headers-aware store nextKeyValue.value is now a
ValueTimestampHeaders<LeftOrRightValue<VLeft, VRight>> (it carries the
headers/timestamp needed to set on the forwarded record), and
forwardNonJoinedOuterRecords consumes that whole wrapper. I can add a named
local of the new type (final ValueTimestampHeaders<LeftOrRightValue<VLeft,
VRight>> outerValue = nextKeyValue.value;) if you'd prefer that for readability
— let me know if the var name is good enough.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.streams.DslStoreFormat;
+import org.apache.kafka.streams.KeyValue;
+import
org.apache.kafka.streams.kstream.internals.AbstractConfigurableStoreFactory;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.internals.StoreFactory;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ValueTimestampHeaders;
+
+/**
+ * Wraps the outer-join store used by {@code KStreamKStreamJoin} so the
processor only deals
+ * with a single value shape: {@code
ValueTimestampHeaders<LeftOrRightValue<VLeft, VRight>>}.
+ * <p>
+ * The underlying store is one of:
+ * <ul>
+ * <li>plain: {@code KeyValueStore<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<VLeft, VRight>>}</li>
+ * <li>headers-aware: {@code KeyValueStore<TimestampedKeyAndJoinSide<K>,
ValueTimestampHeaders<LeftOrRightValue<VLeft, VRight>>>}</li>
+ * </ul>
+ * Both variants are wrapped by the same {@link MeteredKeyValueStore} class —
they only differ
Review Comment:
Mmm. I assume it's not a typo (you did not mean
`MeteredTimestampedKeyValueStoreWithHeaders`) and you are asking for a new
class since if we use `MeteredTimestampedKeyValueStoreWithHeaders`, it
contadicts with your suggestion
[here](https://github.com/apache/kafka/pull/22156/changes#r3365865149)
Do you mean a new type? If yes, we can do it in a follow-up pr?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingListValueBytesStore.java:
##########
@@ -33,9 +32,9 @@ public void put(final Bytes key, final byte[] value) {
// we need to log the full new list and thus call get() on the inner
store below
// if the value is a tombstone, we delete the whole list and thus can
save the get call
if (value == null) {
- log(key, null, internalContext.recordContext().timestamp(), new
RecordHeaders());
+ log(key, null, internalContext.recordContext().timestamp(),
internalContext.recordContext().headers());
} else {
- log(key, wrapped().get(key),
internalContext.recordContext().timestamp(), new RecordHeaders());
+ log(key, wrapped().get(key),
internalContext.recordContext().timestamp(),
internalContext.recordContext().headers());
Review Comment:
What is your main concern? Consistency or correctness? Yes, it;s not
consistent with what we did in
`ChangeLoggingTimestampedKeyValueBytesStoreWithHeaders` but it is still correct.
This store is built with caching always disabled
(`OuterStreamJoinStoreFactory.withCachingDisabled()`), so the layering is just
`MeteredKeyValueStore(ValueTimestampHeadersSerde) ->
ChangeLoggingListValueBytesStore -> ListValueStore`. That means `log(...)` runs
synchronously within the same `process()` call as the put, so
`internalContext.recordContext()` is the record currently being processed —
i.e. the same record whose headers we just stored (`putInOuterJoinStore` passes
`thisRecord.headers()`, which the metered serde writes into the value blob). So
`recordContext().headers()` here is exactly the headers embedded in the value.
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/ListValueStoreTest.java:
##########
@@ -45,37 +49,64 @@
public class ListValueStoreTest {
public enum StoreType { InMemory, RocksDB }
+ public enum ValueMode { Plain, Headers }
+
+ static Stream<Arguments> modes() {
+ return Stream.of(
+ Arguments.of(StoreType.InMemory, ValueMode.Plain),
+ Arguments.of(StoreType.InMemory, ValueMode.Headers),
+ Arguments.of(StoreType.RocksDB, ValueMode.Plain),
+ Arguments.of(StoreType.RocksDB, ValueMode.Headers)
+ );
+ }
- private KeyValueStore<Integer, String> listStore;
+ private StoreFacade facade;
final File baseDir = TestUtils.tempDirectory("test");
- public void setup(final StoreType storeType) {
- listStore = buildStore(Serdes.Integer(), Serdes.String(), storeType);
-
+ public void setup(final StoreType storeType, final ValueMode valueMode) {
final MockRecordCollector recordCollector = new MockRecordCollector();
- final InternalMockProcessorContext<Integer, String> context = new
InternalMockProcessorContext<>(
- baseDir,
- Serdes.String(),
- Serdes.Integer(),
- recordCollector,
- new ThreadCache(
- new LogContext("testCache"),
- 0,
- new MockStreamsMetrics(new Metrics())));
- context.setTime(1L);
-
- listStore.init(context, listStore);
+ final ThreadCache cache = new ThreadCache(
+ new LogContext("testCache"),
+ 0,
+ new MockStreamsMetrics(new Metrics()));
+
+ if (valueMode == ValueMode.Plain) {
+ final KeyValueStore<Integer, String> store =
buildPlainStore(Serdes.Integer(), Serdes.String(), storeType);
+ final InternalMockProcessorContext<Integer, String> context = new
InternalMockProcessorContext<>(
+ baseDir,
+ Serdes.String(),
+ Serdes.Integer(),
+ recordCollector,
+ cache);
+ context.setTime(1L);
+ store.init(context, store);
+ facade = new PlainFacade(store);
+ } else {
+ final KeyValueStore<Integer, ValueTimestampHeaders<String>> store =
+ buildHeadersStore(Serdes.Integer(), Serdes.String(),
storeType);
+ final InternalMockProcessorContext<Integer,
ValueTimestampHeaders<String>> context = new InternalMockProcessorContext<>(
+ baseDir,
+ Serdes.Integer(),
+ null,
Review Comment:
The headers store's value type is ValueTimestampHeaders<String>, for which
there's no off-the-shelf Serde to pass here, and the list store uses its own
serdes from the builder so this default isn't exercised — hence null.
Serdes.Integer() would compile but be misleading, since the value isn't an
Integer (only the key is).
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -248,38 +253,40 @@ private void emitNonJoinedOuterRecords(final
KeyValueStore<TimestampedKeyAndJoin
continue;
}
- final LeftOrRightValue<VLeft, VRight> leftOrRightValue =
nextKeyValue.value;
- forwardNonJoinedOuterRecords(record,
timestampedKeyAndJoinSide, leftOrRightValue);
+ forwardNonJoinedOuterRecords(record,
timestampedKeyAndJoinSide, nextKeyValue.value);
Review Comment:
With the headers-aware store nextKeyValue.value is now a
ValueTimestampHeaders<LeftOrRightValue<VLeft, VRight>> (it carries the
headers/timestamp needed to set on the forwarded record), and
forwardNonJoinedOuterRecords consumes that whole wrapper. I can add a named
local of the new type (final ValueTimestampHeaders<LeftOrRightValue<VLeft,
VRight>> outerValue = nextKeyValue.value;) if you'd prefer that for readability
— let me know if the var name is good enough.
--
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]
