mjsax commented on code in PR #22807:
URL: https://github.com/apache/kafka/pull/22807#discussion_r3591301494
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1348,30 @@ <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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey)}
+ * instead.
*/
+ @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);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.</p>
Review Comment:
```suggestion
* <p>The joiner receives both the mapped join key and the {@link
KStream} record key.
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1367,13 +1385,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
Review Comment:
Why do we remove this? It's still correct?
##########
docs/streams/upgrade-guide.md:
##########
@@ -71,6 +71,8 @@ Kafka Streams no longer emits a WARN from
`KafkaStreams#cleanUp()` when the appl
Kafka Streams now validates the `application.server` configuration when
`StreamsConfig` is created. The value must be empty or a valid endpoint from
which Kafka Streams can parse both host and port, such as `host:port` or
`protocol://host:port`. Invalid values that may previously have failed later
during startup or assignment now fail earlier with a `ConfigException`. More
details can be found in
[KIP-1245](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1245%3A+Enforce+%27application.server%27+%3Cserver%3E%3A%3Cport%3E+format+at+config+level).
+Kafka Streams now exposes the mapped join key alongside the `KStream` record
key in stream-`GlobalKTable` joins, via the new
`ValueJoinerWithStreamAndMappedKey` functional interface. Four new
`KStream#join` and `KStream#leftJoin` overloads accept this joiner, giving
users access to the join key produced by the `KeyValueMapper` in addition to
the stream record's key — without having to recompute the mapped key inside the
joiner. The four existing `ValueJoinerWithKey`-based overloads for
stream-`GlobalKTable` joins are deprecated; their runtime behavior is unchanged
(`readOnlyKey` remains the `KStream` record's key) and applications continue to
compile and run without modification. More details can be found in
[KIP-1340](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1340%3A+Expose+Both+Mapped+Key+and+Stream+Key+in+Streams-GlobalKTable+Joins).
Review Comment:
> stream-`GlobalKTable`
Weird formatting (mix of plain text and class name). Should it be
`KStream`-`GlobalKTable` join? Or "stream-globalTable" join? Two times in the
paragraph.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1480,13 +1534,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
leftJoin(final GlobalKTable<Glob
/**
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
Review Comment:
Same. -- Might be better to keep?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1480,13 +1534,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
leftJoin(final GlobalKTable<Glob
/**
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#leftJoin(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #leftJoin(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 Since 4.4. Use {@link #leftJoin(GlobalKTable,
KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(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,
final Named
named);
+ /**
+ * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}.
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.</p>
Review Comment:
```suggestion
* <p>Takes an additional {@link Named} parameter that is used to name
the processor in the topology.
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1347,13 +1348,30 @@ <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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey)}
+ * instead.
*/
+ @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);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.</p>
Review Comment:
JavaDoc markup is not HTML to we don't need the close tag.
##########
streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+
+/**
+ * The {@code ValueJoinerWithStreamAndMappedKey} 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.
Review Comment:
```suggestion
* arbitrary type, with access to both the original {@link KStream} record
key and the mapped join key.
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1367,13 +1385,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ * <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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @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,
final Named named);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}
Review Comment:
```suggestion
* See {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey, Named)}
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1367,13 +1385,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
join(final GlobalKTable<GlobalKe
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ * <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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @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,
final Named named);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.</p>
Review Comment:
```suggestion
* <p>Takes an additional {@link Named} parameter that is used to name
the processor in the topology.
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java:
##########
@@ -1480,13 +1534,32 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
leftJoin(final GlobalKTable<Glob
/**
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
- * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#leftJoin(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #leftJoin(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 Since 4.4. Use {@link #leftJoin(GlobalKTable,
KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(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,
final Named
named);
+ /**
+ * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}.
Review Comment:
```suggestion
* See {@link #leftJoin(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey, Named)}.
```
--
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]
