KAFKA-20787: Implement KIP-1340: Expose Both Mapped Key and Stream Key in Streams-GlobalKTable Joins

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 `ValueJoinerWithMappedAndStreamKey` 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).
+
 ## Streams API changes in 4.3.0
 
 **Note:** Kafka Streams 4.3.0 contains a critical native memory leak in the RocksDB state store layer ([KAFKA-20616](https://issues.apache.org/jira/browse/KAFKA-20616)). The `ColumnFamilyOptions` for the offsets column family is not closed, and column family handles can leak on close-path exceptions, which under cascading task closes (e.g., rebalances or error-triggered recoveries) leads to unbounded off-heap memory growth and eventual OOM. Users running Kafka Streams should consider upgrading directly to 4.3.1, which includes the fix for it.

streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java

@@ -1345,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.
+     */
+    <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final GlobalKTable<GlobalKey, GlobalValue> globalTable,
+                                                         final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+                                                         final ValueJoinerWithMappedAndStreamKey<? super GlobalKey, ? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner);
+
     /**
      * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
      *
@@ -1365,13 +1381,31 @@ <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 Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)}
+     *             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,
                                                          final Named named);
 
+    /**
+     * As {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}, but the joiner receives both
+     * the mapped join key (used to look up the {@link GlobalKTable} value) and the {@link KStream} record key.
+     */
+    <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final GlobalKTable<GlobalKey, GlobalValue> globalTable,
+                                                         final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+                                                         final ValueJoinerWithMappedAndStreamKey<? super GlobalKey, ? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner,
+                                                         final Named named);
+
     /**
      * Join records of this stream with {@link GlobalKTable}'s records using non-windowed left equi-join.
      * In contrast to an {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner) inner join}, all records from this
@@ -1458,13 +1492,29 @@ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final GlobalKTable<Glob
     /**
      * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}.
      *
-     * <p>Note that the 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 #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 Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)}
+     *             instead, which exposes both the mapped join key and the stream record's key.
      */
+    @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);
 
+    /**
+     * As {@link #leftJoin(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.
+     */
+    <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final GlobalKTable<GlobalKey, GlobalValue> globalTable,
+                                                             final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+                                                             final ValueJoinerWithMappedAndStreamKey<? super GlobalKey, ? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner);
+
     /**
      * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}.
      *
@@ -1478,13 +1528,31 @@ <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 Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)}
+     *             instead, which exposes both the mapped join key and the stream record's key.
      */
+    @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);
 
+    /**
+     * As {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}, but the joiner receives both
+     * the mapped join key (used to look up the {@link GlobalKTable} value) and the {@link KStream} record key.
+     */
+    <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final GlobalKTable<GlobalKey, GlobalValue> globalTable,
+                                                             final KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+                                                             final ValueJoinerWithMappedAndStreamKey<? super GlobalKey, ? super K, ? super V, ? super GlobalValue, ? extends VOut> joiner,
+                                                             final Named named);
+
     /**
      * Process all records in this stream, one record at a time, by applying a {@link Processor} (provided by the given
      * {@link ProcessorSupplier}) to each input record.

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> {
+
+    /**
+     * Return a joined value derived from {@code mappedKey}, {@code streamKey}, {@code value1} and {@code value2}.
+     *
+     * @param mappedKey the join key produced by the {@link KeyValueMapper} (i.e. the {@link GlobalKTable}
+     *                  lookup key); may be {@code null} for a left-join when the mapper returns {@code null}.
+     *                  Read-only.
+     * @param streamKey the {@link KStream} record's key. Read-only.
+     * @param value1    the {@link KStream} record's value
+     * @param value2    the matching {@link GlobalKTable} value, or {@code null} for a left-join with no match
+     * @return the joined value
+     */
+    VR apply(final K1 mappedKey, final K2 streamKey, final V1 value1, final V2 value2);
+}

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;
+    private final KeyValueMapper<? super StreamKey, ? super StreamValue, ? extends TableKey> mapper;
     private final boolean leftJoin;
 
-    KStreamGlobalKTableJoin(final KTableValueGetterSupplier<K2, V2> valueGetterSupplier,
-                            final ValueJoinerWithKey<? super K1, ? super V1, ? super V2, ? extends VOut> joiner,
-                            final KeyValueMapper<? super K1, ? super V1, ? extends K2> mapper,
+    KStreamGlobalKTableJoin(final KTableValueGetterSupplier<TableKey, TableValue> valueGetterSupplier,
+                            final ValueJoinerWithMappedAndStreamKey<? super TableKey, ? super StreamKey, ? super StreamValue, ? super TableValue, ? extends VOut> joiner,
+                            final KeyValueMapper<? super StreamKey, ? super StreamValue, ? extends TableKey> mapper,
                             final boolean leftJoin) {
         this.valueGetterSupplier = valueGetterSupplier;
         this.joiner = joiner;
@@ -41,7 +39,7 @@ class KStreamGlobalKTableJoin<K1, V1, K2, V2, VOut> implements ProcessorSupplier
     }
 
     @Override
-    public Processor<K1, V1, K1, VOut> get() {
-        return new KStreamKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin, Optional.empty(), Optional.empty());
+    public Processor<StreamKey, StreamValue, StreamKey, VOut> get() {
+        return new KStreamGlobalKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin);
     }
 }