root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: remove unnecessary `KeyValueMapper` (#13545) 

Reviewers: Christo Lolov (@clolov), Bill Bejeck <bill@confluent.io>
This commit is contained in:
Matthias J. Sax 2023-04-14 14:37:40 -07:00 committed by GitHub
parent fc9df51be5
commit bf3a5a3e04
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 1 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableKTableInnerJoin.java
View File

@ -17,7 +17,6 @@
package org.apache.kafka.streams.kstream.internals; package org.apache.kafka.streams.kstream.internals;
import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.ValueJoiner; import org.apache.kafka.streams.kstream.ValueJoiner;
import org.apache.kafka.streams.processor.api.ContextualProcessor; import org.apache.kafka.streams.processor.api.ContextualProcessor;
import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Processor;
@ -35,8 +34,6 @@ import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull;
class KTableKTableInnerJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K, V1, V2, VOut> { class KTableKTableInnerJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K, V1, V2, VOut> {
private static final Logger LOG = LoggerFactory.getLogger(KTableKTableInnerJoin.class); private static final Logger LOG = LoggerFactory.getLogger(KTableKTableInnerJoin.class);
private final KeyValueMapper<K, V1, K> keyValueMapper = (key, value) -> key;
KTableKTableInnerJoin(final KTableImpl<K, ?, V1> table1, KTableKTableInnerJoin(final KTableImpl<K, ?, V1> table1,
final KTableImpl<K, ?, V2> table2, final KTableImpl<K, ?, V2> table2,
final ValueJoiner<? super V1, ? super V2, ? extends VOut> joiner) { final ValueJoiner<? super V1, ? super V2, ? extends VOut> joiner) {
@ -172,8 +169,7 @@ class KTableKTableInnerJoin<K, V1, V2, VOut> extends KTableKTableAbstractJoin<K,
final V1 value1 = getValueOrNull(valueAndTimestamp1); final V1 value1 = getValueOrNull(valueAndTimestamp1);
if (value1 != null) { if (value1 != null) {
final ValueAndTimestamp<V2> valueAndTimestamp2 final ValueAndTimestamp<V2> valueAndTimestamp2 = valueGetter2.get(key);
= valueGetter2.get(keyValueMapper.apply(key, value1));
final V2 value2 = getValueOrNull(valueAndTimestamp2); final V2 value2 = getValueOrNull(valueAndTimestamp2);
if (value2 != null) { if (value2 != null) {