mjsax commented on code in PR #22807:
URL: https://github.com/apache/kafka/pull/22807#discussion_r3562667203
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1345,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key — {@link
GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithMappedAndStreamKey)}
Review Comment:
```suggestion
* @deprecated Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithMappedAndStreamKey)}
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey;
+import org.apache.kafka.streams.processor.api.ContextualProcessor;
+import org.apache.kafka.streams.processor.api.ProcessorContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.api.RecordMetadata;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.streams.state.ValueTimestampHeaders;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.kafka.streams.processor.internals.metrics.TaskMetrics.droppedRecordsSensor;
+import static
org.apache.kafka.streams.state.ValueTimestampHeaders.getValueOrNull;
+
+class KStreamGlobalKTableJoinProcessor<StreamKey, StreamValue, TableKey,
TableValue, VOut>
Review Comment:
It seems this new class duplicates `KStreamKTableJoinProcessor` to 99%, with
the key difference of calling `joiner.apply(mappedKey, record.key(),
record.value(), value2)` (different class with an additional argument).
I am wondering, if we should instead change `KStreamKTableJoinProcessor` and
let is use `ValueJoinerWithMappedAndStreamKey`? And for the case when the user
passed `ValueJoinerWithKey`, we wrap it with a
`ValueJoinerWithMappedAndStreamKey` as an "adaptor" which implements
`apply(...)` as
```
VR apply(final K1 mappedKey, final K2 streamKey, final V1 value1, final V2
value2) {
return valueJoinerWithKey.apply(streamKey, value1, value2);
}
```
just dropping the additional `mappedKey` parameter. I believe, we do the
same already to bridge between `ValueJoiner` and `ValueJoinerWithKey` -- uses
can pass either one into a join, and the join `Processor` uses
`ValueJoinerWithKey` so we have a single impl, and a `ValueJoiner` would be
wrapped similarly, dropping the key argument.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java:
##########
@@ -17,22 +17,20 @@
package org.apache.kafka.streams.kstream.internals;
import org.apache.kafka.streams.kstream.KeyValueMapper;
-import org.apache.kafka.streams.kstream.ValueJoinerWithKey;
+import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
-import java.util.Optional;
+class KStreamGlobalKTableJoin<StreamKey, StreamValue, TableKey, TableValue,
VOut> implements ProcessorSupplier<StreamKey, StreamValue, StreamKey, VOut> {
-class KStreamGlobalKTableJoin<K1, V1, K2, V2, VOut> implements
ProcessorSupplier<K1, V1, K1, VOut> {
-
- private final KTableValueGetterSupplier<K2, V2> valueGetterSupplier;
- private final ValueJoinerWithKey<? super K1, ? super V1, ? super V2, ?
extends VOut> joiner;
- private final KeyValueMapper<? super K1, ? super V1, ? extends K2> mapper;
+ private final KTableValueGetterSupplier<TableKey, TableValue>
valueGetterSupplier;
+ private final ValueJoinerWithMappedAndStreamKey<? super TableKey, ? super
StreamKey, ? super StreamValue, ? super TableValue, ? extends VOut> joiner;
Review Comment:
I am just wondering about the order or generic types? As it's a
"stream-globalTable" join, it's a little bit odd that we start with `TableKey`?
I think, we have two options?
- first steam types, second table types: `<StreamKey, StreamValue,
TableKey, TableValue, VOut>`
- first both key, seconds both values: `<StreamKey, TableKey, StreamValue,
TableValue, VOut>`
Thoughts? Sorry, that we missed this during KIP discussion, but easy to
still change and update the KIP.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1345,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key — {@link
GlobalKTable} joins derive
Review Comment:
```suggestion
* joins — where the stream key <em>is</em> the join key —
{@link GlobalKTable} joins derive
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1345,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key — {@link
GlobalKTable} joins derive
Review Comment:
Not sure if an actuallm-dash would render as expected... using html markup
seems to be better? Or remove m-dash and replace with something else.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1345,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key — {@link
GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithMappedAndStreamKey)}
+ * instead, which exposes both the mapped join key and the
stream record's key.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner);
+ /**
+ * As {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}, but the
joiner receives both
+ * the mapped join key (used to look up the {@link GlobalKTable} value)
and the {@link KStream} record key.
Review Comment:
We should try to use a unify JavaDocs style. Existing style is:
```
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
* <maybe explain important differences>
*/
```
We want to avoid duplicating the same description, so we put a full
description into the "simples" variant, and point to overloads. So maybe we
need to update the JavaDoc of `join(GlobalKTable, KeyValueMapper, ValueJoiner)`
which currently point to the now deprecated method and let it point to the new
one?
```
If you need read access to the {@code KStream} key, use {@link
#join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
```
-> change to:
```
If you need read access to the {@code KStream} key or {@link KTable} key
(which is the same as the extract join key), use {@link #join(GlobalKTable,
KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}.
```
Similar for leftJoin() case.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1345,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key — {@link
GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithMappedAndStreamKey)}
+ * instead, which exposes both the mapped join key and the
stream record's key.
Review Comment:
```suggestion
* instead.
```
I would keep it simple
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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.kstream;
+
+/**
+ * The {@code ValueJoinerWithMappedAndStreamKey} interface for joining two
values into a new value of
+ * arbitrary type, with access to both the mapped join key and the original
{@link KStream} record key.
+ * Used by {@link KStream}-{@link GlobalKTable} joins, where the join key is
produced by a
+ * {@link KeyValueMapper} and does not necessarily equal the {@link KStream}
record's key.
+ *
+ * @param <K1> the type of the mapped join key (the {@link GlobalKTable}
lookup key)
+ * @param <K2> the type of the original {@link KStream} record key
+ * @param <V1> the type of the first (stream) value
+ * @param <V2> the type of the second (table) value
+ * @param <VR> the type of the joined result value
+ */
+@FunctionalInterface
+public interface ValueJoinerWithMappedAndStreamKey<K1, K2, V1, V2, VR> {
Review Comment:
Let's use `StreamKey`, `StreamValue` etc similar to other interfaces...
--
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]
