This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8f48a25fdb5 KAFKA-20787: Implement KIP-1340: Expose Both Mapped Key
and Stream Key in Streams-GlobalKTable Joins (#22807)
8f48a25fdb5 is described below
commit 8f48a25fdb5ada256e05569f2276a719895bf644
Author: Lucy Liu <[email protected]>
AuthorDate: Mon Jul 20 11:53:42 2026 -0500
KAFKA-20787: Implement KIP-1340: Expose Both Mapped Key and Stream Key in
Streams-GlobalKTable Joins (#22807)
This PR implements KIP-1340. Prior to this KIP,
`ValueJoinerWithKey`-based KStream-GlobalKTable join pass the `KStream`
record key as the `readOnlyKey`, so users had no way to access the join
key produced by their `KeyValueMapper` inside the joiner without
recomputing it.
This change adds a new joiner interface
`ValueJoinerWithMappedAndStreamKey<K1, K2, V1, V2, VR>`
and new overrides of existing join/leftJoin methods to allow
passing in both the mappedKey and the streamKey.
Reviewers: Matthias J. Sax <[email protected]>
---
docs/streams/upgrade-guide.md | 2 +
.../org/apache/kafka/streams/kstream/KStream.java | 91 +++++++++++-
.../kstream/ValueJoinerWithStreamAndMappedKey.java | 50 +++++++
.../kstream/internals/KStreamGlobalKTableJoin.java | 18 +--
.../streams/kstream/internals/KStreamImpl.java | 48 ++++++
.../kstream/internals/KStreamKTableJoin.java | 8 +-
.../internals/KStreamKTableJoinProcessor.java | 13 +-
.../internals/KStreamGlobalKTableJoinTest.java | 122 ++++++++++++++++
.../internals/KStreamGlobalKTableLeftJoinTest.java | 161 +++++++++++++++++++++
9 files changed, 490 insertions(+), 23 deletions(-)
diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md
index c6e737f56c7..1e1fd54aa21 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+%27applicat
[...]
+Kafka Streams now exposes the mapped join key alongside the `KStream` record
key in KStream-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 KS [...]
+
For applications using the Streams Rebalance Protocol
(`group.protocol=streams`), brokers can now record a human-readable description
of the group's processing topology via a pluggable backend
([KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)).
When the broker configuration
`group.streams.topology.description.plugin.class` is set, Kafka Streams clients
automatically push a description equivalent to `Topology#describe()` t [...]
## 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 900e2eacd99..d532880de90 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
@@ -1282,7 +1282,8 @@ public interface KStream<K, V> {
* <p>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
@@ -1347,13 +1348,30 @@ public interface KStream<K, V> {
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the {@link KStream} key is read-only and must not be
modified, as this can lead to corrupt
- * partitioning and incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key —
{@link GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.
+ */
+ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner);
+
/**
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
@@ -1368,12 +1386,33 @@ public interface KStream<K, V> {
* See {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
* <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ *
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#join(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #join(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key —
{@link GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner,
final Named named);
+ /**
+ * See {@link #join(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey)}
+ *
+ * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ */
+ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner,
+ final Named named);
+
/**
* Join records of this stream with {@link GlobalKTable}'s records using
non-windowed left equi-join.
* In contrast to an {@link #join(GlobalKTable, KeyValueMapper,
ValueJoiner) inner join}, all records from this
@@ -1392,8 +1431,8 @@ public interface KStream<K, V> {
* called with a {@code null} value for the global table record.
* The key of the result record is the same as for both joining input
records,
* or the {@code KStreams} input record's key for a left-join result.
- * If you need read access to the {@code KStream} key, use
- * {@link #leftJoin(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
#leftJoin(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.
* Note, that {@code null} keys for {@code KStream} input records are
supported (in contrast to
@@ -1460,13 +1499,30 @@ public interface KStream<K, V> {
/**
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
- * <p>Note that the key is read-only and must not be modified, as this can
lead to corrupt partitioning and
- * incorrect results.
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#leftJoin(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #leftJoin(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key —
{@link GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Since 4.4. Use {@link #leftJoin(GlobalKTable,
KeyValueMapper, ValueJoinerWithStreamAndMappedKey)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner);
+ /**
+ * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}.
+ *
+ * <p>The joiner receives both the mapped join key and the {@link KStream}
record key.</p>
+ */
+ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner);
+
/**
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}.
*
@@ -1481,12 +1537,33 @@ public interface KStream<K, V> {
* See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)}.
*
* <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ *
+ * <b>Warning:</b> {@code readOnlyKey} is the {@code KStream} record's
key, <b>not</b>
+ * the join key produced by {@code keySelector}. Unlike {@link
#leftJoin(KTable, ValueJoinerWithKey)
+ * KStream-KTable} and {@link #leftJoin(KStream, ValueJoinerWithKey,
JoinWindows) KStream-KStream}
+ * joins — where the stream key <em>is</em> the join key —
{@link GlobalKTable} joins derive
+ * the join key via {@code keySelector}, so {@code readOnlyKey} does
<b>not</b> necessarily
+ * match the key of the {@link GlobalKTable} record being joined.
+ *
+ * @deprecated Since 4.4. Use {@link #leftJoin(GlobalKTable,
KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)}
+ * instead.
*/
+ @Deprecated
<GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner,
final Named
named);
+ /**
+ * See {@link #leftJoin(GlobalKTable, KeyValueMapper,
ValueJoinerWithStreamAndMappedKey)}.
+ *
+ * <p>Takes an additional {@link Named} parameter that is used to name the
processor in the topology.
+ */
+ <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner,
+ final Named
named);
+
/**
* Process all records in this stream, one record at a time, by applying a
{@link Processor} (provided by the given
* {@link ProcessorSupplier}) to each input record.
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
new file mode 100644
index 00000000000..c05dda43545
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithStreamAndMappedKey.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.kstream;
+
+import org.apache.kafka.common.annotation.InterfaceAudience;
+
+/**
+ * The {@code ValueJoinerWithStreamAndMappedKey} interface for joining two
values into a new value of
+ * arbitrary type, with access to both the original {@link KStream} record and
the mapped join 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 <StreamKey> the type of the original {@link KStream} record key
+ * @param <TableKey> the type of the mapped join key (the {@link GlobalKTable}
lookup key)
+ * @param <StreamValue> the type of the first (stream) value
+ * @param <TableValue> the type of the second (table) value
+ * @param <VOut> the type of the joined result value
+ */
+@FunctionalInterface
[email protected]
+public interface ValueJoinerWithStreamAndMappedKey<StreamKey, TableKey,
StreamValue, TableValue, VOut> {
+
+ /**
+ * 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 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
+ */
+ 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 14344b7b43c..97153ad0fb4 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,22 +17,22 @@
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.ValueJoinerWithStreamAndMappedKey;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorSupplier;
import java.util.Optional;
-class KStreamGlobalKTableJoin<K1, V1, K2, V2, VOut> implements
ProcessorSupplier<K1, V1, K1, VOut> {
+class KStreamGlobalKTableJoin<StreamKey, StreamValue, TableKey, TableValue,
VOut> implements ProcessorSupplier<StreamKey, StreamValue, StreamKey, VOut> {
- private final KTableValueGetterSupplier<K2, V2> valueGetterSupplier;
- private final ValueJoinerWithKey<? super K1, ? super V1, ? super V2, ?
extends VOut> joiner;
- private final KeyValueMapper<? super K1, ? super V1, ? extends K2> mapper;
+ private final KTableValueGetterSupplier<TableKey, TableValue>
valueGetterSupplier;
+ private final ValueJoinerWithStreamAndMappedKey<? super StreamKey, ? super
TableKey, ? super StreamValue, ? super TableValue, ? extends VOut> joiner;
+ private final KeyValueMapper<? super StreamKey, ? super StreamValue, ?
extends TableKey> mapper;
private final boolean leftJoin;
- KStreamGlobalKTableJoin(final KTableValueGetterSupplier<K2, V2>
valueGetterSupplier,
- final ValueJoinerWithKey<? super K1, ? super V1, ?
super V2, ? extends VOut> joiner,
- final KeyValueMapper<? super K1, ? super V1, ?
extends K2> mapper,
+ KStreamGlobalKTableJoin(final KTableValueGetterSupplier<TableKey,
TableValue> valueGetterSupplier,
+ final ValueJoinerWithStreamAndMappedKey<? super
StreamKey, ? super TableKey, ? super StreamValue, ? super TableValue, ? extends
VOut> joiner,
+ final KeyValueMapper<? super StreamKey, ? super
StreamValue, ? extends TableKey> mapper,
final boolean leftJoin) {
this.valueGetterSupplier = valueGetterSupplier;
this.joiner = joiner;
@@ -41,7 +41,7 @@ class KStreamGlobalKTableJoin<K1, V1, K2, V2, VOut>
implements ProcessorSupplier
}
@Override
- public Processor<K1, V1, K1, VOut> get() {
+ public Processor<StreamKey, StreamValue, StreamKey, VOut> get() {
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/KStreamImpl.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java
index 02342a9de4a..966c9905a4d 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.Repartitioned;
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.ValueJoinerWithStreamAndMappedKey;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.ValueMapperWithKey;
import org.apache.kafka.streams.kstream.internals.graph.BaseRepartitionNode;
@@ -1185,12 +1186,20 @@ public class KStreamImpl<K, V> extends
AbstractStream<K, V> implements KStream<K
}
@Override
+ @Deprecated
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner) {
return doGlobalTableJoin(globalTable, keySelector, joiner, false,
NamedInternal.empty());
}
+ @Override
+ public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner) {
+ return doGlobalTableJoinWithStreamAndMappedKey(globalTable,
keySelector, joiner, false, NamedInternal.empty());
+ }
+
@Override
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
@@ -1200,6 +1209,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K,
V> implements KStream<K
}
@Override
+ @Deprecated
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner,
@@ -1207,6 +1217,14 @@ public class KStreamImpl<K, V> extends AbstractStream<K,
V> implements KStream<K
return doGlobalTableJoin(globalTable, keySelector, joiner, false,
named);
}
+ @Override
+ public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> join(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner,
+ final Named
named) {
+ return doGlobalTableJoinWithStreamAndMappedKey(globalTable,
keySelector, joiner, false, named);
+ }
+
@Override
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
@@ -1215,12 +1233,20 @@ public class KStreamImpl<K, V> extends
AbstractStream<K, V> implements KStream<K
}
@Override
+ @Deprecated
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner) {
return doGlobalTableJoin(globalTable, keySelector, joiner, true,
NamedInternal.empty());
}
+ @Override
+ public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner) {
+ return doGlobalTableJoinWithStreamAndMappedKey(globalTable,
keySelector, joiner, true, NamedInternal.empty());
+ }
+
@Override
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
@@ -1230,6 +1256,7 @@ public class KStreamImpl<K, V> extends AbstractStream<K,
V> implements KStream<K
}
@Override
+ @Deprecated
public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
final
ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ? extends VOut>
joiner,
@@ -1237,12 +1264,33 @@ public class KStreamImpl<K, V> extends
AbstractStream<K, V> implements KStream<K
return doGlobalTableJoin(globalTable, keySelector, joiner, true,
named);
}
+ @Override
+ public <GlobalKey, GlobalValue, VOut> KStream<K, VOut> leftJoin(final
GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final
KeyValueMapper<? super K, ? super V, ? extends GlobalKey> keySelector,
+ final
ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey, ? super V, ?
super GlobalValue, ? extends VOut> joiner,
+ final
Named named) {
+ return doGlobalTableJoinWithStreamAndMappedKey(globalTable,
keySelector, joiner, true, named);
+ }
+
private <GlobalKey, GlobalValue, VOut> KStream<K, VOut> doGlobalTableJoin(
final GlobalKTable<GlobalKey, GlobalValue> globalTable,
final KeyValueMapper<? super K, ? super V, ? extends GlobalKey>
keySelector,
final ValueJoinerWithKey<? super K, ? super V, ? super GlobalValue, ?
extends VOut> joiner,
final boolean leftJoin,
final Named named
+ ) {
+ Objects.requireNonNull(joiner, "joiner cannot be null");
+ final ValueJoinerWithStreamAndMappedKey<K, GlobalKey, V, GlobalValue,
VOut> adapted =
+ (streamKey, mappedKey, streamValue, tableValue) ->
joiner.apply(streamKey, streamValue, tableValue);
+ return doGlobalTableJoinWithStreamAndMappedKey(globalTable,
keySelector, adapted, leftJoin, named);
+ }
+
+ private <GlobalKey, GlobalValue, VOut> KStream<K, VOut>
doGlobalTableJoinWithStreamAndMappedKey(
+ final GlobalKTable<GlobalKey, GlobalValue> globalTable,
+ final KeyValueMapper<? super K, ? super V, ? extends GlobalKey>
keySelector,
+ final ValueJoinerWithStreamAndMappedKey<? super K, ? super GlobalKey,
? super V, ? super GlobalValue, ? extends VOut> joiner,
+ final boolean leftJoin,
+ final Named named
) {
Objects.requireNonNull(globalTable, "globalTable cannot be null");
Objects.requireNonNull(keySelector, "keySelector cannot be null");
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 dc33a167721..9dd7ac0c4b5 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 @@ class KStreamKTableJoin<StreamKey, StreamValue, TableValue,
VOut> implements Pro
@Override
public Processor<StreamKey, StreamValue, StreamKey, VOut> 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 decbe065211..a1784701eca 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 @@ package org.apache.kafka.streams.kstream.internals;
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<StreamKey, StreamValue,
TableKey, TableValue, V
private final KTableValueGetter<TableKey, TableValue> valueGetter;
private final KeyValueMapper<? super StreamKey, ? super StreamValue, ?
extends TableKey> keyMapper;
- private final ValueJoinerWithKey<? super StreamKey, ? super StreamValue, ?
super TableValue, ? extends VOut> joiner;
+ private final ValueJoinerWithStreamAndMappedKey<? super StreamKey, ? super
TableKey, ? super StreamValue, ? super TableValue, ? extends VOut> joiner;
private final boolean leftJoin;
private Sensor droppedRecordsSensor;
private final Optional<Duration> gracePeriod;
@@ -61,7 +61,7 @@ class KStreamKTableJoinProcessor<StreamKey, StreamValue,
TableKey, TableValue, V
KStreamKTableJoinProcessor(final KTableValueGetter<TableKey, TableValue>
valueGetter,
final KeyValueMapper<? super StreamKey, ? super
StreamValue, ? extends TableKey> keyMapper,
- final ValueJoinerWithKey<? super StreamKey, ?
super StreamValue, ? super TableValue, ? extends VOut> joiner,
+ final ValueJoinerWithStreamAndMappedKey<? super
StreamKey, ? super TableKey, ? super StreamValue, ? super TableValue, ? extends
VOut> joiner,
final boolean leftJoin,
final Optional<Duration> gracePeriod,
final Optional<String> storeName) {
@@ -128,7 +128,7 @@ class KStreamKTableJoinProcessor<StreamKey, StreamValue,
TableKey, TableValue, V
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)));
}
}
@@ -142,8 +142,9 @@ class KStreamKTableJoinProcessor<StreamKey, StreamValue,
TableKey, TableValue, V
private boolean maybeDropRecord(final Record<StreamKey, StreamValue>
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.
+ // 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 6085bdacf68..9260f2f5721 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
@@ -30,6 +30,9 @@ 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;
@@ -50,6 +53,7 @@ import java.util.Set;
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,45 @@ public class KStreamGlobalKTableJoinTest {
}
}
+ private void initWithStreamAndMappedKeyJoiner(
+ final ValueJoinerWithStreamAndMappedKey<Integer, String, String,
String, String> joiner) {
+ driver.close();
+ builder = new StreamsBuilder();
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier =
new MockApiProcessorSupplier<>();
+ final KStream<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+ final KeyValueMapper<Integer, String, String> 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());
+ }
+
+ @SuppressWarnings("deprecation")
+ private void initWithDeprecatedStreamKeyJoiner(
+ final ValueJoinerWithKey<Integer, String, String, String> joiner) {
+ driver.close();
+ builder = new StreamsBuilder();
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier =
new MockApiProcessorSupplier<>();
+ final KStream<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+ final KeyValueMapper<Integer, String, String> 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<Set<String>> copartitionGroups =
@@ -339,4 +382,83 @@ public class KStreamGlobalKTableJoinTest {
is(0.0)
);
}
+
+ @Test
+ public void shouldPassMappedKeyAndStreamKeyToJoiner() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+
+ stream.join(
+ table,
+ (KeyValueMapper<Integer, String, String>) (k, v) -> v,
+ (ValueJoinerWithStreamAndMappedKey<Integer, String, String,
String, String>)
+ (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 6f49917a012..3f5f8365926 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.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;
@@ -50,6 +53,7 @@ import java.util.Set;
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,45 @@ public class KStreamGlobalKTableLeftJoinTest {
}
}
+ private void initWithStreamAndMappedKeyJoiner(
+ final ValueJoinerWithStreamAndMappedKey<Integer, String, String,
String, String> joiner) {
+ driver.close();
+ builder = new StreamsBuilder();
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier =
new MockApiProcessorSupplier<>();
+ final KStream<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+ final KeyValueMapper<Integer, String, String> 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());
+ }
+
+ @SuppressWarnings("deprecation")
+ private void initWithDeprecatedStreamKeyJoiner(
+ final ValueJoinerWithKey<Integer, String, String, String> joiner) {
+ driver.close();
+ builder = new StreamsBuilder();
+ final MockApiProcessorSupplier<Integer, String, Void, Void> supplier =
new MockApiProcessorSupplier<>();
+ final KStream<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+ final KeyValueMapper<Integer, String, String> 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<Set<String>> copartitionGroups =
@@ -349,4 +392,122 @@ public class KStreamGlobalKTableLeftJoinTest {
new KeyValueTimestamp<>(3, "X3,FKey3+Y3", 3)
);
}
+
+ @Test
+ public void shouldPassMappedKeyAndStreamKeyToJoiner() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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() {
+ initWithStreamAndMappedKeyJoiner(
+ (streamKey, mappedKey, 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<Integer, String> stream = builder.stream(streamTopic,
Consumed.with(Serdes.Integer(), Serdes.String()));
+ final GlobalKTable<String, String> table =
builder.globalTable(globalTableTopic, Consumed.with(Serdes.String(),
Serdes.String()));
+
+ stream.leftJoin(
+ table,
+ (KeyValueMapper<Integer, String, String>) (k, v) -> v,
+ (ValueJoinerWithStreamAndMappedKey<Integer, String, String,
String, String>) (mk, sk, v1, v2) -> v1 + v2,
+ Named.as("left-join-table"));
+
+ assertThat(builder.build().describe().toString(),
containsString("Processor: left-join-table"));
+ }
}
