From a1fa6d0134adcdb1bd95c1557b152329c9662e15 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Thu, 4 Jun 2026 12:55:37 -0500 Subject: [PATCH 1/7] validate feasibility of adding ValueJoinerWithKeys new interface --- .../apache/kafka/streams/kstream/KStream.java | 18 +++ .../streams/kstream/ValueJoinerWithKeys.java | 32 +++++ .../internals/KStreamGlobalKTableJoin.java | 10 +- .../KStreamGlobalKTableJoinProcessor.java | 113 ++++++++++++++++++ .../kstream/internals/KStreamImpl.java | 45 ++++++- 5 files changed, 211 insertions(+), 7 deletions(-) create mode 100644 streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java index 5ccc84a12750c..adab78214d7cb 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java @@ -1352,6 +1352,10 @@ KStream join(final GlobalKTable keySelector, final ValueJoinerWithKey joiner); + KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner); + /** * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}. * @@ -1372,6 +1376,11 @@ KStream join(final GlobalKTable joiner, final Named named); + KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys 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 @@ -1465,6 +1474,10 @@ KStream leftJoin(final GlobalKTable keySelector, final ValueJoinerWithKey joiner); + KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner); + /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}. * @@ -1475,6 +1488,11 @@ KStream leftJoin(final GlobalKTable joiner, final Named named); + KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner, + final Named named); + /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}. * diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java new file mode 100644 index 0000000000000..dafcf51226d90 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java @@ -0,0 +1,32 @@ +/* + * 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; + +@FunctionalInterface +public interface ValueJoinerWithKeys { + + /** + * Return a joined value consisting of {@code readOnlyKey}, {@code streamKey}, {@code value1} and {@code value2}. + * + * @param readOnlyKey the readOnly key + * @param streamKey the stream key + * @param value1 the first value for joining + * @param value2 the second value for joining + * @return the joined value + */ + VR apply(final K1 readOnlyKey, final K2 streamKey, final V1 value1, final V2 value2); +} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java index 14344b7b43cef..ba067c4f1a394 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java @@ -17,21 +17,19 @@ 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.ValueJoinerWithKeys; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; -import java.util.Optional; - class KStreamGlobalKTableJoin implements ProcessorSupplier { private final KTableValueGetterSupplier valueGetterSupplier; - private final ValueJoinerWithKey joiner; + private final ValueJoinerWithKeys joiner; private final KeyValueMapper mapper; private final boolean leftJoin; KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, - final ValueJoinerWithKey joiner, + final ValueJoinerWithKeys joiner, final KeyValueMapper mapper, final boolean leftJoin) { this.valueGetterSupplier = valueGetterSupplier; @@ -42,6 +40,6 @@ class KStreamGlobalKTableJoin implements ProcessorSupplier @Override public Processor get() { - return new KStreamKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin, Optional.empty(), Optional.empty()); + return new KStreamGlobalKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java new file mode 100644 index 0000000000000..ff461783ebe15 --- /dev/null +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java @@ -0,0 +1,113 @@ +/* + * 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.ValueJoinerWithKeys; +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 + extends ContextualProcessor { + + private static final Logger LOG = LoggerFactory.getLogger(KStreamGlobalKTableJoinProcessor.class); + + private final KTableValueGetter valueGetter; + private final KeyValueMapper keyMapper; + private final ValueJoinerWithKeys joiner; + private final boolean leftJoin; + private Sensor droppedRecordsSensor; + + KStreamGlobalKTableJoinProcessor(final KTableValueGetter valueGetter, + final KeyValueMapper keyMapper, + final ValueJoinerWithKeys joiner, + final boolean leftJoin) { + this.valueGetter = valueGetter; + this.keyMapper = keyMapper; + this.joiner = joiner; + this.leftJoin = leftJoin; + } + + @Override + public void init(final ProcessorContext context) { + super.init(context); + final StreamsMetricsImpl metrics = (StreamsMetricsImpl) context.metrics(); + droppedRecordsSensor = droppedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics); + valueGetter.init(context); + } + + @Override + public void process(final Record record) { + final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); + if (shouldDrop(record, mappedKey)) { + return; + } + final TableValue tableValue = lookup(record, mappedKey); + if (leftJoin || tableValue != null) { + context().forward(record.withValue(joiner.apply(mappedKey, record.key(), record.value(), tableValue))); + } + } + + private TableValue lookup(final Record record, final TableKey mappedKey) { + if (mappedKey == null) { + return null; + } + final ValueTimestampHeaders valueTimestampHeaders = valueGetter.isVersioned() + ? valueGetter.get(mappedKey, record.timestamp()) + : valueGetter.get(mappedKey); + return getValueOrNull(valueTimestampHeaders); + } + + private boolean shouldDrop(final Record record, final TableKey mappedKey) { + // mirror KStreamKTableJoinProcessor#maybeDropRecord: left-join with null mappedKey but non-null + // value is allowed (produces a null-table-side join result). Drop otherwise. + if (leftJoin && mappedKey == null && record.value() != null) { + return false; + } + if (mappedKey == null || record.value() == null) { + if (context().recordMetadata().isPresent()) { + final RecordMetadata recordMetadata = context().recordMetadata().get(); + LOG.warn( + "Skipping record due to null join key or value. " + + "topic=[{}] partition=[{}] offset=[{}]", + recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset() + ); + } else { + LOG.warn("Skipping record due to null join key or value. Topic, partition, and offset not known."); + } + droppedRecordsSensor.record(); + return true; + } + return false; + } + + @Override + public void close() { + valueGetter.close(); + } +} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java index 02342a9de4a75..e6e3c32d78fb0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java @@ -39,6 +39,7 @@ import org.apache.kafka.streams.kstream.StreamJoined; import org.apache.kafka.streams.kstream.ValueJoiner; import org.apache.kafka.streams.kstream.ValueJoinerWithKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithKeys; import org.apache.kafka.streams.kstream.ValueMapper; import org.apache.kafka.streams.kstream.ValueMapperWithKey; import org.apache.kafka.streams.kstream.internals.graph.BaseRepartitionNode; @@ -1191,6 +1192,13 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner) { + return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, false, NamedInternal.empty()); + } + @Override public KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1207,6 +1215,14 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner, + final Named named) { + return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, false, named); + } + @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1221,6 +1237,13 @@ public KStream leftJoin(final GlobalKTab return doGlobalTableJoin(globalTable, keySelector, joiner, true, NamedInternal.empty()); } + @Override + public KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner) { + return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, true, NamedInternal.empty()); + } + @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1237,12 +1260,33 @@ public KStream leftJoin(final GlobalKTab return doGlobalTableJoin(globalTable, keySelector, joiner, true, named); } + @Override + public KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner, + final Named named) { + return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, true, named); + } + private KStream doGlobalTableJoin( final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, final boolean leftJoin, final Named named + ) { + Objects.requireNonNull(joiner, "joiner cannot be null"); + final ValueJoinerWithKeys adapted = + (mappedKey, streamKey, streamValue, tableValue) -> joiner.apply(streamKey, streamValue, tableValue); + return doGlobalTableJoinWithKeys(globalTable, keySelector, adapted, leftJoin, named); + } + + private KStream doGlobalTableJoinWithKeys( + final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithKeys joiner, + final boolean leftJoin, + final Named named ) { Objects.requireNonNull(globalTable, "globalTable cannot be null"); Objects.requireNonNull(keySelector, "keySelector cannot be null"); @@ -1266,7 +1310,6 @@ private KStream doGlobalTableJoin( } builder.addGraphNode(graphNode, streamTableJoinNode); - // do not have serde for joined result return new KStreamImpl<>( name, keySerde, From 2bde80c9749780e42cd0f83a4c48a202e5b3e4cb Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Thu, 4 Jun 2026 14:16:39 -0500 Subject: [PATCH 2/7] make naming adjustments --- .../apache/kafka/streams/kstream/KStream.java | 12 ++++++++ .../internals/KStreamGlobalKTableJoin.java | 16 +++++----- .../KStreamGlobalKTableJoinProcessor.java | 29 ++++++++++--------- .../kstream/internals/KStreamImpl.java | 5 ++++ 4 files changed, 41 insertions(+), 21 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java index adab78214d7cb..d1c8cc85ddcee 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java @@ -1347,7 +1347,10 @@ KStream join(final GlobalKTableNote that the {@link KStream} key is read-only and must not be modified, as this can lead to corrupt * partitioning and incorrect results. + * + * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. */ + @Deprecated KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner); @@ -1370,7 +1373,10 @@ KStream join(final GlobalKTableTakes an additional {@link Named} parameter that is used to name the processor in the topology. + * + * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys, Named)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. */ + @Deprecated KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, @@ -1469,7 +1475,10 @@ KStream leftJoin(final GlobalKTableNote that the key is read-only and must not be modified, as this can lead to corrupt partitioning and * incorrect results. + * + * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. */ + @Deprecated KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner); @@ -1497,7 +1506,10 @@ KStream leftJoin(final GlobalKTableTakes an additional {@link Named} parameter that is used to name the processor in the topology. + * + * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys, Named)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. */ + @Deprecated KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java index ba067c4f1a394..ba561d6f2978b 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java @@ -21,16 +21,16 @@ import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; -class KStreamGlobalKTableJoin implements ProcessorSupplier { +class KStreamGlobalKTableJoin implements ProcessorSupplier { - private final KTableValueGetterSupplier valueGetterSupplier; - private final ValueJoinerWithKeys joiner; - private final KeyValueMapper mapper; + private final KTableValueGetterSupplier valueGetterSupplier; + private final ValueJoinerWithKeys joiner; + private final KeyValueMapper mapper; private final boolean leftJoin; - KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, - final ValueJoinerWithKeys joiner, - final KeyValueMapper mapper, + KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, + final ValueJoinerWithKeys joiner, + final KeyValueMapper mapper, final boolean leftJoin) { this.valueGetterSupplier = valueGetterSupplier; this.joiner = joiner; @@ -39,7 +39,7 @@ class KStreamGlobalKTableJoin implements ProcessorSupplier } @Override - public Processor get() { + public Processor get() { return new KStreamGlobalKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java index ff461783ebe15..f83d7c996b007 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java @@ -63,29 +63,30 @@ public void init(final ProcessorContext context) { @Override public void process(final Record record) { - final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); - if (shouldDrop(record, mappedKey)) { + if (maybeDropRecord(record)) { return; } - final TableValue tableValue = lookup(record, mappedKey); - if (leftJoin || tableValue != null) { - context().forward(record.withValue(joiner.apply(mappedKey, record.key(), record.value(), tableValue))); - } + doJoin(record); } - private TableValue lookup(final Record record, final TableKey mappedKey) { - if (mappedKey == null) { - return null; + private void doJoin(final Record record) { + final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); + final TableValue value2 = getTableValue(record, mappedKey); + if (leftJoin || value2 != null) { + context().forward(record.withValue(joiner.apply(mappedKey, record.key(), record.value(), value2))); } + } + + private TableValue getTableValue(final Record record, final TableKey mappedKey) { + if (mappedKey == null) return null; final ValueTimestampHeaders valueTimestampHeaders = valueGetter.isVersioned() ? valueGetter.get(mappedKey, record.timestamp()) : valueGetter.get(mappedKey); return getValueOrNull(valueTimestampHeaders); } - private boolean shouldDrop(final Record record, final TableKey mappedKey) { - // mirror KStreamKTableJoinProcessor#maybeDropRecord: left-join with null mappedKey but non-null - // value is allowed (produces a null-table-side join result). Drop otherwise. + private boolean maybeDropRecord(final Record record) { + final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); if (leftJoin && mappedKey == null && record.value() != null) { return false; } @@ -98,7 +99,9 @@ private boolean shouldDrop(final Record record, final Ta recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset() ); } else { - LOG.warn("Skipping record due to null join key or value. Topic, partition, and offset not known."); + LOG.warn( + "Skipping record due to null join key or value. Topic, partition, and offset not known." + ); } droppedRecordsSensor.record(); return true; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java index e6e3c32d78fb0..fa6332671dd3e 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java @@ -1186,6 +1186,7 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner) { @@ -1208,6 +1209,7 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, @@ -1231,6 +1233,7 @@ public KStream leftJoin(final GlobalKTab } @Override + @Deprecated public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner) { @@ -1253,6 +1256,7 @@ public KStream leftJoin(final GlobalKTab } @Override + @Deprecated public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, @@ -1310,6 +1314,7 @@ private KStream doGlobalTableJoinWithKey } builder.addGraphNode(graphNode, streamTableJoinNode); + // do not have serde for joined result return new KStreamImpl<>( name, keySerde, From 45c1ae6ada4c8c8af2cad49b6bc6d78e86ccd64e Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 6 Jul 2026 11:10:00 -0500 Subject: [PATCH 3/7] add functionality --- .../apache/kafka/streams/kstream/KStream.java | 74 ++++++++++++++----- .../streams/kstream/ValueJoinerWithKeys.java | 32 -------- .../ValueJoinerWithMappedAndStreamKey.java | 46 ++++++++++++ .../internals/KStreamGlobalKTableJoin.java | 6 +- .../KStreamGlobalKTableJoinProcessor.java | 17 ++--- .../kstream/internals/KStreamImpl.java | 26 +++---- 6 files changed, 126 insertions(+), 75 deletions(-) delete mode 100644 streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java create mode 100644 streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java index d1c8cc85ddcee..f3af51fd0d24a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java @@ -1345,19 +1345,28 @@ KStream join(final GlobalKTableNote that the {@link KStream} key is read-only and must not be modified, as this can lead to corrupt - * partitioning and incorrect results. + * Warning: {@code readOnlyKey} is the {@code KStream} record's key, not + * 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 is the join key — {@link GlobalKTable} joins derive + * the join key via {@code keySelector}, so {@code readOnlyKey} does not necessarily + * match the key of the {@link GlobalKTable} record being joined. * - * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. + * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)} + * instead, which exposes both the mapped join key and the stream record's key. */ @Deprecated KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey 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. + */ KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner); + final ValueJoinerWithMappedAndStreamKey joiner); /** * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}. @@ -1372,9 +1381,15 @@ KStream join(final GlobalKTableTakes an additional {@link Named} parameter that is used to name the processor in the topology. + * Warning: {@code readOnlyKey} is the {@code KStream} record's key, not + * 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 is the join key — {@link GlobalKTable} joins derive + * the join key via {@code keySelector}, so {@code readOnlyKey} does not necessarily + * match the key of the {@link GlobalKTable} record being joined. * - * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys, Named)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. + * @deprecated Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)} + * instead, which exposes both the mapped join key and the stream record's key. */ @Deprecated KStream join(final GlobalKTable globalTable, @@ -1382,9 +1397,13 @@ KStream join(final GlobalKTable 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. + */ KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final Named named); /** @@ -1473,19 +1492,28 @@ KStream leftJoin(final GlobalKTableNote that the key is read-only and must not be modified, as this can lead to corrupt partitioning and - * incorrect results. + * Warning: {@code readOnlyKey} is the {@code KStream} record's key, not + * 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 is the join key — {@link GlobalKTable} joins derive + * the join key via {@code keySelector}, so {@code readOnlyKey} does not necessarily + * match the key of the {@link GlobalKTable} record being joined. * - * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. + * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey)} + * instead, which exposes both the mapped join key and the stream record's key. */ @Deprecated KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey 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. + */ KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner); + final ValueJoinerWithMappedAndStreamKey joiner); /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}. @@ -1497,17 +1525,18 @@ KStream leftJoin(final GlobalKTable joiner, final Named named); - KStream leftJoin(final GlobalKTable globalTable, - final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner, - final Named named); - /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}. * - *

Takes an additional {@link Named} parameter that is used to name the processor in the topology. + * Warning: {@code readOnlyKey} is the {@code KStream} record's key, not + * 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 is the join key — {@link GlobalKTable} joins derive + * the join key via {@code keySelector}, so {@code readOnlyKey} does not necessarily + * match the key of the {@link GlobalKTable} record being joined. * - * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKeys, Named)} instead. This overload passes the {@link KStream} record's key as {@code readOnlyKey}, not the join key produced by {@code keySelector}. + * @deprecated Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithMappedAndStreamKey, Named)} + * instead, which exposes both the mapped join key and the stream record's key. */ @Deprecated KStream leftJoin(final GlobalKTable globalTable, @@ -1515,6 +1544,15 @@ KStream leftJoin(final GlobalKTable 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. + */ + KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithMappedAndStreamKey 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. diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java deleted file mode 100644 index dafcf51226d90..0000000000000 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKeys.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * 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; - -@FunctionalInterface -public interface ValueJoinerWithKeys { - - /** - * Return a joined value consisting of {@code readOnlyKey}, {@code streamKey}, {@code value1} and {@code value2}. - * - * @param readOnlyKey the readOnly key - * @param streamKey the stream key - * @param value1 the first value for joining - * @param value2 the second value for joining - * @return the joined value - */ - VR apply(final K1 readOnlyKey, final K2 streamKey, final V1 value1, final V2 value2); -} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java new file mode 100644 index 0000000000000..e9ea1d7575b97 --- /dev/null +++ b/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 the type of the mapped join key (the {@link GlobalKTable} lookup key) + * @param the type of the original {@link KStream} record key + * @param the type of the first (stream) value + * @param the type of the second (table) value + * @param the type of the joined result value + */ +@FunctionalInterface +public interface ValueJoinerWithMappedAndStreamKey { + + /** + * Return a joined value consisting of {@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); +} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java index ba561d6f2978b..dda820e780db9 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java @@ -17,19 +17,19 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.streams.kstream.KeyValueMapper; -import org.apache.kafka.streams.kstream.ValueJoinerWithKeys; +import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; class KStreamGlobalKTableJoin implements ProcessorSupplier { private final KTableValueGetterSupplier valueGetterSupplier; - private final ValueJoinerWithKeys joiner; + private final ValueJoinerWithMappedAndStreamKey joiner; private final KeyValueMapper mapper; private final boolean leftJoin; KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final KeyValueMapper mapper, final boolean leftJoin) { this.valueGetterSupplier = valueGetterSupplier; diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java index f83d7c996b007..995709cbc13b6 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java @@ -18,7 +18,7 @@ import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.streams.kstream.KeyValueMapper; -import org.apache.kafka.streams.kstream.ValueJoinerWithKeys; +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; @@ -39,13 +39,13 @@ class KStreamGlobalKTableJoinProcessor valueGetter; private final KeyValueMapper keyMapper; - private final ValueJoinerWithKeys joiner; + private final ValueJoinerWithMappedAndStreamKey joiner; private final boolean leftJoin; private Sensor droppedRecordsSensor; KStreamGlobalKTableJoinProcessor(final KTableValueGetter valueGetter, final KeyValueMapper keyMapper, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final boolean leftJoin) { this.valueGetter = valueGetter; this.keyMapper = keyMapper; @@ -63,14 +63,14 @@ public void init(final ProcessorContext context) { @Override public void process(final Record record) { - if (maybeDropRecord(record)) { + final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); + if (maybeDropRecord(record, mappedKey)) { return; } - doJoin(record); + doJoin(record, mappedKey); } - private void doJoin(final Record record) { - final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); + private void doJoin(final Record record, final TableKey mappedKey) { final TableValue value2 = getTableValue(record, mappedKey); if (leftJoin || value2 != null) { context().forward(record.withValue(joiner.apply(mappedKey, record.key(), record.value(), value2))); @@ -85,8 +85,7 @@ private TableValue getTableValue(final Record record, fi return getValueOrNull(valueTimestampHeaders); } - private boolean maybeDropRecord(final Record record) { - final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); + private boolean maybeDropRecord(final Record record, final TableKey mappedKey) { if (leftJoin && mappedKey == null && record.value() != null) { return false; } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java index fa6332671dd3e..6a98c9413c0a3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java @@ -39,7 +39,7 @@ import org.apache.kafka.streams.kstream.StreamJoined; import org.apache.kafka.streams.kstream.ValueJoiner; import org.apache.kafka.streams.kstream.ValueJoinerWithKey; -import org.apache.kafka.streams.kstream.ValueJoinerWithKeys; +import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; import org.apache.kafka.streams.kstream.ValueMapper; import org.apache.kafka.streams.kstream.ValueMapperWithKey; import org.apache.kafka.streams.kstream.internals.graph.BaseRepartitionNode; @@ -1196,8 +1196,8 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner) { - return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, false, NamedInternal.empty()); + final ValueJoinerWithMappedAndStreamKey joiner) { + return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, false, NamedInternal.empty()); } @Override @@ -1220,9 +1220,9 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final Named named) { - return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, false, named); + return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, false, named); } @Override @@ -1243,8 +1243,8 @@ public KStream leftJoin(final GlobalKTab @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner) { - return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, true, NamedInternal.empty()); + final ValueJoinerWithMappedAndStreamKey joiner) { + return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, true, NamedInternal.empty()); } @Override @@ -1267,9 +1267,9 @@ public KStream leftJoin(final GlobalKTab @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final Named named) { - return doGlobalTableJoinWithKeys(globalTable, keySelector, joiner, true, named); + return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, true, named); } private KStream doGlobalTableJoin( @@ -1280,15 +1280,15 @@ private KStream doGlobalTableJoin( final Named named ) { Objects.requireNonNull(joiner, "joiner cannot be null"); - final ValueJoinerWithKeys adapted = + final ValueJoinerWithMappedAndStreamKey adapted = (mappedKey, streamKey, streamValue, tableValue) -> joiner.apply(streamKey, streamValue, tableValue); - return doGlobalTableJoinWithKeys(globalTable, keySelector, adapted, leftJoin, named); + return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, adapted, leftJoin, named); } - private KStream doGlobalTableJoinWithKeys( + private KStream doGlobalTableJoinWithMappedAndStreamKey( final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithKeys joiner, + final ValueJoinerWithMappedAndStreamKey joiner, final boolean leftJoin, final Named named ) { From 6bb8b10814ef56f10eca565f0368c42c871b787d Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Wed, 8 Jul 2026 15:59:50 -0500 Subject: [PATCH 4/7] add streams tests --- .../internals/KStreamKTableJoinProcessor.java | 4 +- .../KStreamGlobalKTableJoinTest.java | 124 +++++++++++++- .../KStreamGlobalKTableLeftJoinTest.java | 160 ++++++++++++++++++ 3 files changed, 281 insertions(+), 7 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java index decbe0652115e..45451c0e3f7bd 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java @@ -142,8 +142,8 @@ private TableValue getTableValue(final Record record, fi private boolean maybeDropRecord(final Record record) { // we do join iff the join keys are equal, thus, if {@code keyMapper} returns {@code null} we - // cannot join and just ignore the record. Note for KTables, this is the same as having a null key - // since keyMapper just returns the key, but for GlobalKTables we can have other keyMappers + // cannot join and just ignore the record. For stream-KTable joins the {@code keyMapper} is the + // identity, so a {@code null} mapped key is equivalent to a {@code null} stream record key. // // we also ignore the record if value is null, because in a key-value data model a null-value indicates // an empty message (ie, there is nothing to be joined) -- this contrast SQL NULL semantics diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java index 6085bdacf68f8..aa090ca52c748 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java @@ -25,11 +25,7 @@ import org.apache.kafka.streams.TestInputTopic; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.TopologyWrapper; -import org.apache.kafka.streams.kstream.Consumed; -import org.apache.kafka.streams.kstream.GlobalKTable; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KeyValueMapper; -import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.*; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.test.MockApiProcessor; import org.apache.kafka.test.MockApiProcessorSupplier; @@ -50,6 +46,7 @@ import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -141,6 +138,44 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } + private void initWithMappedAndStreamKeyJoiner( + final ValueJoinerWithMappedAndStreamKey joiner) { + driver.close(); + builder = new StreamsBuilder(); + final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + final KeyValueMapper keyMapper = (key, value) -> { + if (value == null) return null; + final String[] tokens = value.split(","); + return tokens.length > 1 ? tokens[1] : null; + }; + stream.join(table, keyMapper, joiner).process(supplier); + driver = new TopologyTestDriver(builder.build(), StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String())); + processor = supplier.theCapturedProcessor(); + inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ofMillis(1L)); + inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); + } + + private void initWithDeprecatedStreamKeyJoiner( + final ValueJoinerWithKey joiner) { + driver.close(); + builder = new StreamsBuilder(); + final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + final KeyValueMapper keyMapper = (key, value) -> { + if (value == null) return null; + final String[] tokens = value.split(","); + return tokens.length > 1 ? tokens[1] : null; + }; + stream.join(table, keyMapper, joiner).process(supplier); + driver = new TopologyTestDriver(builder.build(), StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String())); + processor = supplier.theCapturedProcessor(); + inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ofMillis(1L)); + inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); + } + @Test public void shouldNotRequireCopartitioning() { final Collection> copartitionGroups = @@ -339,4 +374,83 @@ public void shouldJoinOnNullKey() { is(0.0) ); } + + @Test + public void shouldPassMappedKeyAndStreamKeyToJoiner() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + + pushToGlobalTable(2, "Y"); + pushToStream(2, "X", true, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "FKey0|0|X0,FKey0|Y0", 0), + new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1) + ); + } + + @Test + public void shouldDropRecordAndRecordDroppedSensorWhenStreamValueIsNull() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + pushToGlobalTable(2, "Y"); + inputStreamTopic.pipeInput(0, null); + inputStreamTopic.pipeInput(1, "X1,FKey1"); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1) + ); + + assertThat( + driver.metrics().get( + new MetricName( + "dropped-records-total", + "stream-task-metrics", + "", + mkMap( + mkEntry("thread-id", Thread.currentThread().getName()), + mkEntry("task-id", "0_0") + ) + )) + .metricValue(), + is(1.0) + ); + } + + @Test + public void shouldNameProcessorBasedOnNamedParameter() { + final StreamsBuilder builder = new StreamsBuilder(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + + stream.join( + table, + (KeyValueMapper) (k, v) -> v, + (ValueJoinerWithMappedAndStreamKey) + (mk, sk, v1, v2) -> v1 + v2, + Named.as("join-table")); + assertThat(builder.build().describe().toString(), containsString("Processor: join-table")); + } + + /** + * Test the situation when the deprecated {@link ValueJoinerWithKey} overload of + * stream-globalTable join is used. The joiner should receive the {@link KStream} + * record's key as {@code readOnlyKey} only. + */ + @SuppressWarnings("deprecation") + @Test + public void shouldPassStreamKeyAsReadOnlyKeyToDeprecatedJoiner() { + initWithDeprecatedStreamKeyJoiner( + (readOnlyKey, streamValue, tableValue) -> readOnlyKey + "|" + streamValue + "|" + tableValue); + + pushToGlobalTable(2, "Y"); + pushToStream(2, "X", true, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "0|X0,FKey0|Y0", 0), + new KeyValueTimestamp<>(1, "1|X1,FKey1|Y1", 1) + ); + } } diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java index 6f49917a012f9..b2f0c5e819e9a 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java @@ -30,6 +30,9 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KeyValueMapper; import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Named; +import org.apache.kafka.streams.kstream.ValueJoinerWithKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.test.MockApiProcessor; import org.apache.kafka.test.MockApiProcessorSupplier; @@ -50,6 +53,7 @@ import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -141,6 +145,44 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } + private void initWithMappedAndStreamKeyJoiner( + final ValueJoinerWithMappedAndStreamKey joiner) { + driver.close(); + builder = new StreamsBuilder(); + final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + final KeyValueMapper keyMapper = (key, value) -> { + if (value == null) return null; + final String[] tokens = value.split(","); + return tokens.length > 1 ? tokens[1] : null; + }; + stream.leftJoin(table, keyMapper, joiner).process(supplier); + driver = new TopologyTestDriver(builder.build(), StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String())); + processor = supplier.theCapturedProcessor(); + inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ofMillis(1L)); + inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); + } + + private void initWithDeprecatedStreamKeyJoiner( + final ValueJoinerWithKey joiner) { + driver.close(); + builder = new StreamsBuilder(); + final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + final KeyValueMapper keyMapper = (key, value) -> { + if (value == null) return null; + final String[] tokens = value.split(","); + return tokens.length > 1 ? tokens[1] : null; + }; + stream.leftJoin(table, keyMapper, joiner).process(supplier); + driver = new TopologyTestDriver(builder.build(), StreamsTestUtils.getStreamsConfig(Serdes.Integer(), Serdes.String())); + processor = supplier.theCapturedProcessor(); + inputStreamTopic = driver.createInputTopic(streamTopic, new IntegerSerializer(), new StringSerializer(), Instant.ofEpochMilli(0L), Duration.ofMillis(1L)); + inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); + } + @Test public void shouldNotRequireCopartitioning() { final Collection> copartitionGroups = @@ -349,4 +391,122 @@ public void shouldJoinOnNullKey() { new KeyValueTimestamp<>(3, "X3,FKey3+Y3", 3) ); } + + @Test + public void shouldPassMappedKeyAndStreamKeyToJoiner() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + + pushToGlobalTable(2, "Y"); + pushToStream(2, "X", true, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "FKey0|0|X0,FKey0|Y0", 0), + new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1) + ); + } + + @Test + public void shouldDropRecordAndRecordDroppedSensorWhenStreamValueIsNull() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + pushToGlobalTable(2, "Y"); + inputStreamTopic.pipeInput(0, null); + inputStreamTopic.pipeInput(1, "X1,FKey1"); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|Y1", 1) + ); + + assertThat( + driver.metrics().get( + new MetricName( + "dropped-records-total", + "stream-task-metrics", + "", + mkMap( + mkEntry("thread-id", Thread.currentThread().getName()), + mkEntry("task-id", "0_0") + ) + )) + .metricValue(), + is(1.0) + ); + } + + @SuppressWarnings("deprecation") + @Test + public void shouldPassStreamKeyAsReadOnlyKeyToDeprecatedJoiner() { + initWithDeprecatedStreamKeyJoiner( + (readOnlyKey, streamValue, tableValue) -> readOnlyKey + "|" + streamValue + "|" + tableValue); + + pushToGlobalTable(2, "Y"); + pushToStream(2, "X", true, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "0|X0,FKey0|Y0", 0), + new KeyValueTimestamp<>(1, "1|X1,FKey1|Y1", 1) + ); + } + + @Test + public void shouldEmitWithNullTableValueOnNoMatch() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + + pushToStream(2, "X", true, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "FKey0|0|X0,FKey0|null", 0), + new KeyValueTimestamp<>(1, "FKey1|1|X1,FKey1|null", 1) + ); + } + + @Test + public void shouldEmitWithNullMappedKeyWhenMapperReturnsNull() { + initWithMappedAndStreamKeyJoiner( + (mappedKey, streamKey, streamValue, tableValue) -> + mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); + + pushToGlobalTable(2, "Y"); + pushToStream(2, "XXX", false, false); + + processor.checkAndClearProcessResult( + new KeyValueTimestamp<>(0, "null|0|XXX0|null", 0), + new KeyValueTimestamp<>(1, "null|1|XXX1|null", 1) + ); + + assertThat( + driver.metrics().get( + new MetricName( + "dropped-records-total", + "stream-task-metrics", + "", + mkMap( + mkEntry("thread-id", Thread.currentThread().getName()), + mkEntry("task-id", "0_0") + ) + )) + .metricValue(), + is(0.0) + ); + } + + @Test + public void shouldNameProcessorBasedOnNamedParameter() { + final StreamsBuilder builder = new StreamsBuilder(); + final KStream stream = builder.stream(streamTopic, Consumed.with(Serdes.Integer(), Serdes.String())); + final GlobalKTable table = builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(), Serdes.String())); + + stream.leftJoin( + table, + (KeyValueMapper) (k, v) -> v, + (ValueJoinerWithMappedAndStreamKey) (mk, sk, v1, v2) -> v1 + v2, + Named.as("left-join-table")); + + assertThat(builder.build().describe().toString(), containsString("Processor: left-join-table")); + } } From 67f8b1dc80f1ae5695b68170bc544bf8fa6f8969 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Fri, 10 Jul 2026 16:09:56 -0500 Subject: [PATCH 5/7] add upgrade guides --- docs/streams/upgrade-guide.md | 2 ++ .../streams/kstream/ValueJoinerWithMappedAndStreamKey.java | 2 +- .../streams/kstream/internals/KStreamGlobalKTableJoinTest.java | 1 + .../kstream/internals/KStreamGlobalKTableLeftJoinTest.java | 1 + 4 files changed, 5 insertions(+), 1 deletion(-) diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md index df78b93d72c5f..3c1b808ecb3f9 100644 --- a/docs/streams/upgrade-guide.md +++ b/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. diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java index e9ea1d7575b97..338f7ac3e6c54 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java @@ -32,7 +32,7 @@ public interface ValueJoinerWithMappedAndStreamKey { /** - * Return a joined value consisting of {@code mappedKey}, {@code streamKey}, {@code value1} and {@code value2}. + * 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}. diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java index aa090ca52c748..53b8f9c73e0b6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java @@ -157,6 +157,7 @@ private void initWithMappedAndStreamKeyJoiner( inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); } + @SuppressWarnings("deprecation") private void initWithDeprecatedStreamKeyJoiner( final ValueJoinerWithKey joiner) { driver.close(); diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java index b2f0c5e819e9a..326beb63cd075 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java @@ -164,6 +164,7 @@ private void initWithMappedAndStreamKeyJoiner( inputTableTopic = driver.createInputTopic(globalTableTopic, new StringSerializer(), new StringSerializer()); } + @SuppressWarnings("deprecation") private void initWithDeprecatedStreamKeyJoiner( final ValueJoinerWithKey joiner) { driver.close(); From 48ac1099ac6d33d842ed5900bc66826738265454 Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 13 Jul 2026 14:53:57 -0500 Subject: [PATCH 6/7] change param order, remove KStreamGlobalKTableJoinProcessor and address nits --- docs/streams/upgrade-guide.md | 2 +- .../apache/kafka/streams/kstream/KStream.java | 59 +++++---- ...=> ValueJoinerWithStreamAndMappedKey.java} | 19 +-- .../internals/KStreamGlobalKTableJoin.java | 10 +- .../KStreamGlobalKTableJoinProcessor.java | 115 ------------------ .../kstream/internals/KStreamImpl.java | 28 ++--- .../kstream/internals/KStreamKTableJoin.java | 8 +- .../internals/KStreamKTableJoinProcessor.java | 9 +- .../KStreamGlobalKTableJoinTest.java | 23 ++-- .../KStreamGlobalKTableLeftJoinTest.java | 24 ++-- 10 files changed, 102 insertions(+), 195 deletions(-) rename streams/src/main/java/org/apache/kafka/streams/kstream/{ValueJoinerWithMappedAndStreamKey.java => ValueJoinerWithStreamAndMappedKey.java} (74%) delete mode 100644 streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md index 3c1b808ecb3f9..87b167e941c18 100644 --- a/docs/streams/upgrade-guide.md +++ b/docs/streams/upgrade-guide.md @@ -71,7 +71,7 @@ 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). +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). ## Streams API changes in 4.3.0 diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java index f3af51fd0d24a..cda79cc1192e4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java @@ -1280,7 +1280,8 @@ KStream leftJoin(final KTable table, *

For each {@code KStream} record that finds a joining record in the {@link GlobalKTable} the provided * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record. * The key of the result record is the same as the stream record's key. - * If you need read access to the {@code KStream} key, use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}. + * If you need read access to the {@code KStream} key or the {@link GlobalKTable} key (which is the same as the + * join key extracted by {@code keySelector}), use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey)}. * If a {@code KStream} input record's value is {@code null} or if the provided {@link KeyValueMapper keySelector} * returns {@code null}, the input record will be dropped, and no join computation is triggered. * If a {@link GlobalKTable} input record's key is {@code null} the input record will be dropped, and the table @@ -1348,12 +1349,12 @@ KStream join(final GlobalKTableWarning: {@code readOnlyKey} is the {@code KStream} record's key, not * 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 is the join key — {@link GlobalKTable} joins derive + * joins — where the stream key is the join key — {@link GlobalKTable} joins derive * the join key via {@code keySelector}, so {@code readOnlyKey} does not 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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey)} + * instead. */ @Deprecated KStream join(final GlobalKTable globalTable, @@ -1361,12 +1362,13 @@ KStream join(final GlobalKTable 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. + * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}. + * + *

The joiner receives both the mapped join key and the {@link KStream} record key.

*/ KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner); + final ValueJoinerWithStreamAndMappedKey joiner); /** * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}. @@ -1384,12 +1386,12 @@ KStream join(final GlobalKTableWarning: {@code readOnlyKey} is the {@code KStream} record's key, not * 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 is the join key — {@link GlobalKTable} joins derive + * joins — where the stream key is the join key — {@link GlobalKTable} joins derive * the join key via {@code keySelector}, so {@code readOnlyKey} does not 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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)} + * instead. */ @Deprecated KStream join(final GlobalKTable globalTable, @@ -1398,12 +1400,13 @@ KStream join(final GlobalKTableThe joiner receives both the mapped join key and the {@link KStream} record key.

*/ KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final Named named); /** @@ -1424,8 +1427,8 @@ KStream join(final GlobalKTable KStream leftJoin(final GlobalKTableWarning: {@code readOnlyKey} is the {@code KStream} record's key, not * 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 is the join key — {@link GlobalKTable} joins derive + * joins — where the stream key is the join key — {@link GlobalKTable} joins derive * the join key via {@code keySelector}, so {@code readOnlyKey} does not 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 Since 4.4. Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey)} + * instead. */ @Deprecated KStream leftJoin(final GlobalKTable globalTable, @@ -1508,12 +1511,13 @@ KStream leftJoin(final GlobalKTable 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. + * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}. + * + *

The joiner receives both the mapped join key and the {@link KStream} record key.

*/ KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner); + final ValueJoinerWithStreamAndMappedKey joiner); /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}. @@ -1531,12 +1535,12 @@ KStream leftJoin(final GlobalKTableWarning: {@code readOnlyKey} is the {@code KStream} record's key, not * 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 is the join key — {@link GlobalKTable} joins derive + * joins — where the stream key is the join key — {@link GlobalKTable} joins derive * the join key via {@code keySelector}, so {@code readOnlyKey} does not 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 Since 4.4. Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)} + * instead. */ @Deprecated KStream leftJoin(final GlobalKTable globalTable, @@ -1545,12 +1549,13 @@ KStream leftJoin(final GlobalKTableThe joiner receives both the mapped join key and the {@link KStream} record key.

*/ KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final Named named); /** diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java similarity index 74% rename from streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java rename to streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java index 338f7ac3e6c54..960d0e2d2be90 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithMappedAndStreamKey.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java @@ -17,30 +17,31 @@ package org.apache.kafka.streams.kstream; /** - * The {@code ValueJoinerWithMappedAndStreamKey} interface for joining two values into a new value of + * 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. * 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 the type of the mapped join key (the {@link GlobalKTable} lookup key) - * @param the type of the original {@link KStream} record key - * @param the type of the first (stream) value - * @param the type of the second (table) value - * @param the type of the joined result value + * + * @param the type of the original {@link KStream} record key + * @param the type of the mapped join key (the {@link GlobalKTable} lookup key) + * @param the type of the first (stream) value + * @param the type of the second (table) value + * @param the type of the joined result value */ @FunctionalInterface -public interface ValueJoinerWithMappedAndStreamKey { +public interface ValueJoinerWithStreamAndMappedKey { /** * Return a joined value derived from {@code mappedKey}, {@code streamKey}, {@code value1} and {@code value2}. * + * @param streamKey the {@link KStream} record's key. Read-only. * @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); + VOut apply(final StreamKey streamKey, final TableKey mappedKey, final StreamValue value1, final TableValue value2); } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java index dda820e780db9..97153ad0fb4cf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoin.java @@ -17,19 +17,21 @@ package org.apache.kafka.streams.kstream.internals; import org.apache.kafka.streams.kstream.KeyValueMapper; -import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithStreamAndMappedKey; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.ProcessorSupplier; +import java.util.Optional; + class KStreamGlobalKTableJoin implements ProcessorSupplier { private final KTableValueGetterSupplier valueGetterSupplier; - private final ValueJoinerWithMappedAndStreamKey joiner; + private final ValueJoinerWithStreamAndMappedKey joiner; private final KeyValueMapper mapper; private final boolean leftJoin; KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final KeyValueMapper mapper, final boolean leftJoin) { this.valueGetterSupplier = valueGetterSupplier; @@ -40,6 +42,6 @@ class KStreamGlobalKTableJoin get() { - return new KStreamGlobalKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin); + return new KStreamKTableJoinProcessor<>(valueGetterSupplier.get(), mapper, joiner, leftJoin, Optional.empty(), Optional.empty()); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java deleted file mode 100644 index 995709cbc13b6..0000000000000 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinProcessor.java +++ /dev/null @@ -1,115 +0,0 @@ -/* - * 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 - extends ContextualProcessor { - - private static final Logger LOG = LoggerFactory.getLogger(KStreamGlobalKTableJoinProcessor.class); - - private final KTableValueGetter valueGetter; - private final KeyValueMapper keyMapper; - private final ValueJoinerWithMappedAndStreamKey joiner; - private final boolean leftJoin; - private Sensor droppedRecordsSensor; - - KStreamGlobalKTableJoinProcessor(final KTableValueGetter valueGetter, - final KeyValueMapper keyMapper, - final ValueJoinerWithMappedAndStreamKey joiner, - final boolean leftJoin) { - this.valueGetter = valueGetter; - this.keyMapper = keyMapper; - this.joiner = joiner; - this.leftJoin = leftJoin; - } - - @Override - public void init(final ProcessorContext context) { - super.init(context); - final StreamsMetricsImpl metrics = (StreamsMetricsImpl) context.metrics(); - droppedRecordsSensor = droppedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics); - valueGetter.init(context); - } - - @Override - public void process(final Record record) { - final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); - if (maybeDropRecord(record, mappedKey)) { - return; - } - doJoin(record, mappedKey); - } - - private void doJoin(final Record record, final TableKey mappedKey) { - final TableValue value2 = getTableValue(record, mappedKey); - if (leftJoin || value2 != null) { - context().forward(record.withValue(joiner.apply(mappedKey, record.key(), record.value(), value2))); - } - } - - private TableValue getTableValue(final Record record, final TableKey mappedKey) { - if (mappedKey == null) return null; - final ValueTimestampHeaders valueTimestampHeaders = valueGetter.isVersioned() - ? valueGetter.get(mappedKey, record.timestamp()) - : valueGetter.get(mappedKey); - return getValueOrNull(valueTimestampHeaders); - } - - private boolean maybeDropRecord(final Record record, final TableKey mappedKey) { - if (leftJoin && mappedKey == null && record.value() != null) { - return false; - } - if (mappedKey == null || record.value() == null) { - if (context().recordMetadata().isPresent()) { - final RecordMetadata recordMetadata = context().recordMetadata().get(); - LOG.warn( - "Skipping record due to null join key or value. " - + "topic=[{}] partition=[{}] offset=[{}]", - recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset() - ); - } else { - LOG.warn( - "Skipping record due to null join key or value. Topic, partition, and offset not known." - ); - } - droppedRecordsSensor.record(); - return true; - } - return false; - } - - @Override - public void close() { - valueGetter.close(); - } -} diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java index 6a98c9413c0a3..966c9905a4df4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java @@ -39,7 +39,7 @@ import org.apache.kafka.streams.kstream.StreamJoined; import org.apache.kafka.streams.kstream.ValueJoiner; import org.apache.kafka.streams.kstream.ValueJoinerWithKey; -import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithStreamAndMappedKey; import org.apache.kafka.streams.kstream.ValueMapper; import org.apache.kafka.streams.kstream.ValueMapperWithKey; import org.apache.kafka.streams.kstream.internals.graph.BaseRepartitionNode; @@ -1196,8 +1196,8 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner) { - return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, false, NamedInternal.empty()); + final ValueJoinerWithStreamAndMappedKey joiner) { + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, false, NamedInternal.empty()); } @Override @@ -1220,9 +1220,9 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final Named named) { - return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, false, named); + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, false, named); } @Override @@ -1243,8 +1243,8 @@ public KStream leftJoin(final GlobalKTab @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner) { - return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, true, NamedInternal.empty()); + final ValueJoinerWithStreamAndMappedKey joiner) { + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, true, NamedInternal.empty()); } @Override @@ -1267,9 +1267,9 @@ public KStream leftJoin(final GlobalKTab @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final Named named) { - return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, joiner, true, named); + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, true, named); } private KStream doGlobalTableJoin( @@ -1280,15 +1280,15 @@ private KStream doGlobalTableJoin( final Named named ) { Objects.requireNonNull(joiner, "joiner cannot be null"); - final ValueJoinerWithMappedAndStreamKey adapted = - (mappedKey, streamKey, streamValue, tableValue) -> joiner.apply(streamKey, streamValue, tableValue); - return doGlobalTableJoinWithMappedAndStreamKey(globalTable, keySelector, adapted, leftJoin, named); + final ValueJoinerWithStreamAndMappedKey adapted = + (streamKey, mappedKey, streamValue, tableValue) -> joiner.apply(streamKey, streamValue, tableValue); + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, adapted, leftJoin, named); } - private KStream doGlobalTableJoinWithMappedAndStreamKey( + private KStream doGlobalTableJoinWithStreamAndMappedKey( final GlobalKTable globalTable, final KeyValueMapper keySelector, - final ValueJoinerWithMappedAndStreamKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final boolean leftJoin, final Named named ) { diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java index dc33a167721ca..9dd7ac0c4b5c4 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java @@ -58,7 +58,13 @@ public Set> stores() { @Override public Processor get() { - return new KStreamKTableJoinProcessor<>(valueGetterSupplier.get(), keyValueMapper, joiner, leftJoin, gracePeriod, storeName); + return new KStreamKTableJoinProcessor<>( + valueGetterSupplier.get(), + keyValueMapper, + (streamKey, mappedKey, value1, value2) -> joiner.apply(streamKey, value1, value2), + leftJoin, + gracePeriod, + storeName); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java index 45451c0e3f7bd..a1784701eca4f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java @@ -19,7 +19,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.streams.kstream.KeyValueMapper; -import org.apache.kafka.streams.kstream.ValueJoinerWithKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithStreamAndMappedKey; import org.apache.kafka.streams.processor.api.ContextualProcessor; import org.apache.kafka.streams.processor.api.ProcessorContext; import org.apache.kafka.streams.processor.api.Record; @@ -49,7 +49,7 @@ class KStreamKTableJoinProcessor valueGetter; private final KeyValueMapper keyMapper; - private final ValueJoinerWithKey joiner; + private final ValueJoinerWithStreamAndMappedKey joiner; private final boolean leftJoin; private Sensor droppedRecordsSensor; private final Optional gracePeriod; @@ -61,7 +61,7 @@ class KStreamKTableJoinProcessor valueGetter, final KeyValueMapper keyMapper, - final ValueJoinerWithKey joiner, + final ValueJoinerWithStreamAndMappedKey joiner, final boolean leftJoin, final Optional gracePeriod, final Optional storeName) { @@ -128,7 +128,7 @@ private void doJoin(final Record record) { final TableKey mappedKey = keyMapper.apply(record.key(), record.value()); final TableValue value2 = getTableValue(record, mappedKey); if (leftJoin || value2 != null) { - internalProcessorContext.forward(record.withValue(joiner.apply(record.key(), record.value(), value2))); + internalProcessorContext.forward(record.withValue(joiner.apply(record.key(), mappedKey, record.value(), value2))); } } @@ -144,6 +144,7 @@ private boolean maybeDropRecord(final Record record) { // we do join iff the join keys are equal, thus, if {@code keyMapper} returns {@code null} we // cannot join and just ignore the record. For stream-KTable joins the {@code keyMapper} is the // identity, so a {@code null} mapped key is equivalent to a {@code null} stream record key. + // For stream-GlobalKTable joins, {@code keyMapper} is the user-supplied mapper instead. // // we also ignore the record if value is null, because in a key-value data model a null-value indicates // an empty message (ie, there is nothing to be joined) -- this contrast SQL NULL semantics diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java index 53b8f9c73e0b6..9260f2f5721f3 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableJoinTest.java @@ -25,7 +25,14 @@ import org.apache.kafka.streams.TestInputTopic; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.TopologyWrapper; -import org.apache.kafka.streams.kstream.*; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.kstream.GlobalKTable; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Named; +import org.apache.kafka.streams.kstream.ValueJoinerWithKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithStreamAndMappedKey; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.test.MockApiProcessor; import org.apache.kafka.test.MockApiProcessorSupplier; @@ -138,8 +145,8 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } - private void initWithMappedAndStreamKeyJoiner( - final ValueJoinerWithMappedAndStreamKey joiner) { + private void initWithStreamAndMappedKeyJoiner( + final ValueJoinerWithStreamAndMappedKey joiner) { driver.close(); builder = new StreamsBuilder(); final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); @@ -378,8 +385,8 @@ public void shouldJoinOnNullKey() { @Test public void shouldPassMappedKeyAndStreamKeyToJoiner() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToGlobalTable(2, "Y"); @@ -393,8 +400,8 @@ public void shouldPassMappedKeyAndStreamKeyToJoiner() { @Test public void shouldDropRecordAndRecordDroppedSensorWhenStreamValueIsNull() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToGlobalTable(2, "Y"); inputStreamTopic.pipeInput(0, null); @@ -429,7 +436,7 @@ public void shouldNameProcessorBasedOnNamedParameter() { stream.join( table, (KeyValueMapper) (k, v) -> v, - (ValueJoinerWithMappedAndStreamKey) + (ValueJoinerWithStreamAndMappedKey) (mk, sk, v1, v2) -> v1 + v2, Named.as("join-table")); assertThat(builder.build().describe().toString(), containsString("Processor: join-table")); diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java index 326beb63cd075..3f5f8365926e0 100644 --- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamGlobalKTableLeftJoinTest.java @@ -32,7 +32,7 @@ import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Named; import org.apache.kafka.streams.kstream.ValueJoinerWithKey; -import org.apache.kafka.streams.kstream.ValueJoinerWithMappedAndStreamKey; +import org.apache.kafka.streams.kstream.ValueJoinerWithStreamAndMappedKey; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.test.MockApiProcessor; import org.apache.kafka.test.MockApiProcessorSupplier; @@ -145,8 +145,8 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } - private void initWithMappedAndStreamKeyJoiner( - final ValueJoinerWithMappedAndStreamKey joiner) { + private void initWithStreamAndMappedKeyJoiner( + final ValueJoinerWithStreamAndMappedKey joiner) { driver.close(); builder = new StreamsBuilder(); final MockApiProcessorSupplier supplier = new MockApiProcessorSupplier<>(); @@ -395,8 +395,8 @@ public void shouldJoinOnNullKey() { @Test public void shouldPassMappedKeyAndStreamKeyToJoiner() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToGlobalTable(2, "Y"); @@ -410,8 +410,8 @@ public void shouldPassMappedKeyAndStreamKeyToJoiner() { @Test public void shouldDropRecordAndRecordDroppedSensorWhenStreamValueIsNull() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToGlobalTable(2, "Y"); inputStreamTopic.pipeInput(0, null); @@ -454,8 +454,8 @@ public void shouldPassStreamKeyAsReadOnlyKeyToDeprecatedJoiner() { @Test public void shouldEmitWithNullTableValueOnNoMatch() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToStream(2, "X", true, false); @@ -468,8 +468,8 @@ public void shouldEmitWithNullTableValueOnNoMatch() { @Test public void shouldEmitWithNullMappedKeyWhenMapperReturnsNull() { - initWithMappedAndStreamKeyJoiner( - (mappedKey, streamKey, streamValue, tableValue) -> + initWithStreamAndMappedKeyJoiner( + (streamKey, mappedKey, streamValue, tableValue) -> mappedKey + "|" + streamKey + "|" + streamValue + "|" + tableValue); pushToGlobalTable(2, "Y"); @@ -505,7 +505,7 @@ public void shouldNameProcessorBasedOnNamedParameter() { stream.leftJoin( table, (KeyValueMapper) (k, v) -> v, - (ValueJoinerWithMappedAndStreamKey) (mk, sk, v1, v2) -> v1 + v2, + (ValueJoinerWithStreamAndMappedKey) (mk, sk, v1, v2) -> v1 + v2, Named.as("left-join-table")); assertThat(builder.build().describe().toString(), containsString("Processor: left-join-table")); From e4c93e2d1923e29f7bf6141490e23386938395eb Mon Sep 17 00:00:00 2001 From: lucliu1108 Date: Mon, 13 Jul 2026 15:30:34 -0500 Subject: [PATCH 7/7] add InterfaceAudience.Public annotation --- .../streams/kstream/ValueJoinerWithStreamAndMappedKey.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java index 960d0e2d2be90..01c948e88b6a3 100644 --- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java +++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java @@ -16,6 +16,8 @@ */ 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. @@ -30,6 +32,7 @@ * @param the type of the joined result value */ @FunctionalInterface +@InterfaceAudience.Public public interface ValueJoinerWithStreamAndMappedKey { /**