mjsax commented on code in PR #22156:
URL: https://github.com/apache/kafka/pull/22156#discussion_r3591569627
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -221,13 +226,14 @@ private void emitNonJoinedOuterRecords(final
KeyValueStore<TimestampedKeyAndJoin
// reset to MAX_VALUE in case the store is empty
sharedTimeTracker.minTime = Long.MAX_VALUE;
- try (final KeyValueIterator<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<VLeft, VRight>> it = store.all()) {
+ final OuterJoinStoreWrapper<K, VLeft, VRight> wrapper =
outerJoinStoreWrapper.get();
Review Comment:
Does this result in a compiler warning about an Optional.get access w/o
`isPresent()` check? I believe, the goal of passing in the store as parameter
into this method was, to avoid such a warning.
Not a big deal, works either way, but I believe the previous pattern (even
if not obvious at first glance) is the cleaner one?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -299,27 +308,28 @@ private long getOuterJoinLookBackTimeMs(
private void emitInnerJoin(final Record<K, VThis> thisRecord, final
KeyValue<Long, VOther> otherRecord,
final long inputRecordTimestamp) {
- outerJoinStore.ifPresent(store -> {
+ if (outerJoinStoreWrapper.isPresent()) {
// use putIfAbsent to first read and see if there's any values
for the key,
// if yes delete the key, otherwise do not issue a put;
// we may delete some values with the same key early but since
we are going
// range over all values of the same key even after failure,
since the other window-store
// is only cleaned up by stream time, so this is okay for
at-least-once.
final TimestampedKeyAndJoinSide<K> otherKey =
makeOtherKey(thisRecord.key(), otherRecord.key);
- store.putIfAbsent(otherKey, null);
- });
+ outerJoinStoreWrapper.get().putIfAbsent(otherKey, null, null);
+ }
context().forward(
thisRecord.withValue(joiner.apply(thisRecord.key(),
thisRecord.value(), otherRecord.value))
.withTimestamp(Math.max(inputRecordTimestamp,
otherRecord.key)));
}
private void putInOuterJoinStore(final Record<K, VThis> thisRecord) {
- outerJoinStore.ifPresent(store -> {
- final TimestampedKeyAndJoinSide<K> thisKey =
makeThisKey(thisRecord.key(), thisRecord.timestamp());
- final LeftOrRightValue<VLeft, VRight> thisValue =
makeThisValue(thisRecord.value());
- store.put(thisKey, thisValue);
- });
+ if (outerJoinStoreWrapper.isEmpty()) {
Review Comment:
Same
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -299,27 +308,28 @@ private long getOuterJoinLookBackTimeMs(
private void emitInnerJoin(final Record<K, VThis> thisRecord, final
KeyValue<Long, VOther> otherRecord,
final long inputRecordTimestamp) {
- outerJoinStore.ifPresent(store -> {
+ if (outerJoinStoreWrapper.isPresent()) {
Review Comment:
Similar to my other comment. I believe (?) idiomatic Java would use
`ifPreset` to access the store from the `Optional` and get it passed as lambda
parameter, avoiding the `get()` call later.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/OuterJoinStoreWrapper.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.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+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.AggregationWithHeaders;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.KeyValueStore;
+
+/**
+ * Wraps the outer-join store used by {@code KStreamKStreamJoin} so the
processor only deals
+ * with a single value shape: {@code
AggregationWithHeaders<LeftOrRightValue<VLeft, VRight>>}.
+ * <p>
+ * The value carries no timestamp: the entry timestamp is part of the key
+ * ({@link TimestampedKeyAndJoinSide#timestamp()}), which is what the entries
are sorted by, so a
+ * separate value-side timestamp would be redundant.
+ * <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>,
AggregationWithHeaders<LeftOrRightValue<VLeft, VRight>>>}</li>
+ * </ul>
+ * Both variants are wrapped by the same {@link MeteredKeyValueStore} class —
they only differ
+ * in the value serde and (erased) generic parameter — so the variant cannot
be detected by
+ * casting the runtime store. Instead, the variant is read from the {@link
StoreFactory}'s
+ * configured {@link DslStoreFormat}.
+ */
+public class OuterJoinStoreWrapper<K, VLeft, VRight> {
+
+ private final boolean isHeadersStore;
+ private KeyValueStore<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<VLeft, VRight>> plainStore;
+ private KeyValueStore<TimestampedKeyAndJoinSide<K>,
AggregationWithHeaders<LeftOrRightValue<VLeft, VRight>>> headersStore;
+
+ public OuterJoinStoreWrapper(final ProcessorContext<?, ?> context, final
StoreFactory storeFactory) {
+ this.isHeadersStore = isHeadersAware(storeFactory);
+ if (isHeadersStore) {
+ headersStore = context.getStateStore(storeFactory.storeName());
+ } else {
+ plainStore = context.getStateStore(storeFactory.storeName());
+ }
+ }
+
+ private static boolean isHeadersAware(final StoreFactory storeFactory) {
+ if (storeFactory instanceof AbstractConfigurableStoreFactory) {
+ return ((AbstractConfigurableStoreFactory)
storeFactory).dslStoreFormat() == DslStoreFormat.HEADERS;
+ }
+ return false;
+ }
+
+ public boolean isHeadersStore() {
+ return isHeadersStore;
+ }
+
+ public void put(final TimestampedKeyAndJoinSide<K> key,
+ final LeftOrRightValue<VLeft, VRight> value,
+ final Headers headers) {
+ if (headersStore != null) {
+ headersStore.put(key, value == null
+ ? null
+ : AggregationWithHeaders.makeAllowNullable(value, headers));
Review Comment:
Why don't we just call `make(...)` -- if does handle `value == null` case
already correctly I believe?
Similar further below.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/OuterStreamJoinStoreFactory.java:
##########
@@ -96,7 +94,9 @@ public StoreBuilder<?> builder() {
final TimestampedKeyAndJoinSideSerde<K> timestampedKeyAndJoinSideSerde
= new TimestampedKeyAndJoinSideSerde<>(streamJoined.keySerde());
final LeftOrRightValueSerde<V1, V2> leftOrRightValueSerde = new
LeftOrRightValueSerde<>(streamJoined.valueSerde(),
streamJoined.otherValueSerde());
- // Once the headers-aware version of ListValueStore is implemented
(planned for AK 4.4), replace the PLAIN constant with the dslStoreFormat()
method.
+ // We always use the PLAIN underlying bytes-store supplier: the
per-element headers
+ // (when in HEADERS mode) are embedded inside each list element's
value bytes, so
+ // the bytes-store layer itself needs no special header support.
Review Comment:
Wondering if this is actually safe from an upgrade POV? If we have a
pre-header list-store, it would use a PLAIN store, and thus put all the data
into default-CL using "plain value serde".
If we upgrade to header store, don't we need to use format HEADERS, so we
get a `RocksDBTimestampedStoreWithHeaders` which can still read from the
default CL and inserting empty headers on read, but writing new records into
the headers-CF?
If we use a PLAIN store, and pass in `AggregationWithHeadersSerde`, I
believe it won't be able to correctly read the old "plain value format"?
--
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]
