diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md index df78b93d72c5f..87b167e941c18 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 `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 **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/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java index 900e2eacd9933..e030055974a1d 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 @@ 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 @@ -1347,13 +1348,30 @@ 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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey)} + * instead. */ + @Deprecated KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner); + /** + * 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 ValueJoinerWithStreamAndMappedKey joiner); + /** * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner)}. * @@ -1367,13 +1385,32 @@ 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 Since 4.4. Use {@link #join(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)} + * instead. */ + @Deprecated KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, final Named named); + /** + * See {@link #join(GlobalKTable, KeyValueMapper, ValueJoiner, Named)} + * + *

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

+ */ + KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey 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 +1429,8 @@ KStream join(final GlobalKTable 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 Since 4.4. Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey)} + * instead. */ + @Deprecated KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner); + /** + * 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 ValueJoinerWithStreamAndMappedKey joiner); + /** * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)}. * @@ -1480,13 +1534,32 @@ KStream leftJoin(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 #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 Since 4.4. Use {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithStreamAndMappedKey, Named)} + * instead. */ + @Deprecated KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, final Named named); + /** + * See {@link #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner, Named)}. + * + *

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

+ */ + KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey 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 0000000000000..01c948e88b6a3 --- /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 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 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 +@InterfaceAudience.Public +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 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 14344b7b43cef..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,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 implements ProcessorSupplier { +class KStreamGlobalKTableJoin implements ProcessorSupplier { - private final KTableValueGetterSupplier valueGetterSupplier; - private final ValueJoinerWithKey joiner; - private final KeyValueMapper mapper; + private final KTableValueGetterSupplier valueGetterSupplier; + private final ValueJoinerWithStreamAndMappedKey joiner; + private final KeyValueMapper mapper; private final boolean leftJoin; - KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, - final ValueJoinerWithKey joiner, - final KeyValueMapper mapper, + KStreamGlobalKTableJoin(final KTableValueGetterSupplier valueGetterSupplier, + final ValueJoinerWithStreamAndMappedKey joiner, + final KeyValueMapper mapper, final boolean leftJoin) { this.valueGetterSupplier = valueGetterSupplier; this.joiner = joiner; @@ -41,7 +41,7 @@ class KStreamGlobalKTableJoin implements ProcessorSupplier } @Override - public Processor get() { + public Processor 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 02342a9de4a75..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,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.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 KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner) { return doGlobalTableJoin(globalTable, keySelector, joiner, false, NamedInternal.empty()); } + @Override + public KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey joiner) { + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, false, NamedInternal.empty()); + } + @Override public KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1200,6 +1209,7 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, @@ -1207,6 +1217,14 @@ public KStream join(final GlobalKTable KStream join(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey joiner, + final Named named) { + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, false, named); + } + @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1215,12 +1233,20 @@ public KStream leftJoin(final GlobalKTab } @Override + @Deprecated public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner) { return doGlobalTableJoin(globalTable, keySelector, joiner, true, NamedInternal.empty()); } + @Override + public KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey joiner) { + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, joiner, true, NamedInternal.empty()); + } + @Override public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, @@ -1230,6 +1256,7 @@ public KStream leftJoin(final GlobalKTab } @Override + @Deprecated public KStream leftJoin(final GlobalKTable globalTable, final KeyValueMapper keySelector, final ValueJoinerWithKey joiner, @@ -1237,12 +1264,33 @@ public KStream leftJoin(final GlobalKTab return doGlobalTableJoin(globalTable, keySelector, joiner, true, named); } + @Override + public KStream leftJoin(final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey joiner, + final Named named) { + return doGlobalTableJoinWithStreamAndMappedKey(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 ValueJoinerWithStreamAndMappedKey adapted = + (streamKey, mappedKey, streamValue, tableValue) -> joiner.apply(streamKey, streamValue, tableValue); + return doGlobalTableJoinWithStreamAndMappedKey(globalTable, keySelector, adapted, leftJoin, named); + } + + private KStream doGlobalTableJoinWithStreamAndMappedKey( + final GlobalKTable globalTable, + final KeyValueMapper keySelector, + final ValueJoinerWithStreamAndMappedKey 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 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 decbe0652115e..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))); } } @@ -142,8 +142,9 @@ 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. + // 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 6085bdacf68f8..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 @@ -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.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 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 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } + private void initWithStreamAndMappedKeyJoiner( + final ValueJoinerWithStreamAndMappedKey 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()); + } + + @SuppressWarnings("deprecation") + 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 +382,83 @@ public void shouldJoinOnNullKey() { 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 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, + (ValueJoinerWithStreamAndMappedKey) + (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..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 @@ -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.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 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 @@ private void pushNullValueToGlobalTable(final int messageCount) { } } + private void initWithStreamAndMappedKeyJoiner( + final ValueJoinerWithStreamAndMappedKey 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()); + } + + @SuppressWarnings("deprecation") + 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 +392,122 @@ public void shouldJoinOnNullKey() { 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 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, + (ValueJoinerWithStreamAndMappedKey) (mk, sk, v1, v2) -> v1 + v2, + Named.as("left-join-table")); + + assertThat(builder.build().describe().toString(), containsString("Processor: left-join-table")); + } }