Browse Source
Implements KIP-862: https://cwiki.apache.org/confluence/x/WSf1D Reviewers: Guozhang Wang <guozhang@apache.org>, Austin Heyne <aheyne>, John Roesler <vvcephei@apache.org>pull/12720/head
Vicky Papavasileiou
2 years ago
committed by
GitHub
11 changed files with 1103 additions and 39 deletions
@ -0,0 +1,142 @@
@@ -0,0 +1,142 @@
|
||||
/* |
||||
* 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 static org.apache.kafka.streams.processor.internals.metrics.TaskMetrics.droppedRecordsSensor; |
||||
|
||||
import org.apache.kafka.common.metrics.Sensor; |
||||
import org.apache.kafka.streams.KeyValue; |
||||
import org.apache.kafka.streams.kstream.ValueJoinerWithKey; |
||||
import org.apache.kafka.streams.kstream.internals.KStreamImplJoin.TimeTracker; |
||||
import org.apache.kafka.streams.processor.api.ContextualProcessor; |
||||
import org.apache.kafka.streams.processor.api.Processor; |
||||
import org.apache.kafka.streams.processor.api.ProcessorContext; |
||||
import org.apache.kafka.streams.processor.api.ProcessorSupplier; |
||||
import org.apache.kafka.streams.processor.api.Record; |
||||
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; |
||||
import org.apache.kafka.streams.state.WindowStore; |
||||
import org.apache.kafka.streams.state.WindowStoreIterator; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
class KStreamKStreamSelfJoin<K, V1, V2, VOut> implements ProcessorSupplier<K, V1, K, VOut> { |
||||
private static final Logger LOG = LoggerFactory.getLogger(KStreamKStreamSelfJoin.class); |
||||
|
||||
private final String windowName; |
||||
private final long joinThisBeforeMs; |
||||
private final long joinThisAfterMs; |
||||
private final long joinOtherBeforeMs; |
||||
private final long joinOtherAfterMs; |
||||
private final ValueJoinerWithKey<? super K, ? super V1, ? super V2, ? extends VOut> joinerThis; |
||||
|
||||
private final TimeTracker sharedTimeTracker; |
||||
|
||||
KStreamKStreamSelfJoin( |
||||
final String windowName, |
||||
final JoinWindowsInternal windows, |
||||
final ValueJoinerWithKey<? super K, ? super V1, ? super V2, ? extends VOut> joinerThis, |
||||
final TimeTracker sharedTimeTracker) { |
||||
|
||||
this.windowName = windowName; |
||||
this.joinThisBeforeMs = windows.beforeMs; |
||||
this.joinThisAfterMs = windows.afterMs; |
||||
this.joinOtherBeforeMs = windows.afterMs; |
||||
this.joinOtherAfterMs = windows.beforeMs; |
||||
this.joinerThis = joinerThis; |
||||
this.sharedTimeTracker = sharedTimeTracker; |
||||
} |
||||
|
||||
@Override |
||||
public Processor<K, V1, K, VOut> get() { |
||||
return new KStreamKStreamSelfJoinProcessor(); |
||||
} |
||||
|
||||
private class KStreamKStreamSelfJoinProcessor extends ContextualProcessor<K, V1, K, VOut> { |
||||
private WindowStore<K, V2> windowStore; |
||||
private Sensor droppedRecordsSensor; |
||||
|
||||
@Override |
||||
public void init(final ProcessorContext<K, VOut> context) { |
||||
super.init(context); |
||||
|
||||
final StreamsMetricsImpl metrics = (StreamsMetricsImpl) context.metrics(); |
||||
droppedRecordsSensor = droppedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics); |
||||
windowStore = context.getStateStore(windowName); |
||||
} |
||||
|
||||
@SuppressWarnings("unchecked") |
||||
@Override |
||||
public void process(final Record<K, V1> record) { |
||||
if (StreamStreamJoinUtil.skipRecord(record, LOG, droppedRecordsSensor, context())) { |
||||
return; |
||||
} |
||||
|
||||
final long inputRecordTimestamp = record.timestamp(); |
||||
long timeFrom = Math.max(0L, inputRecordTimestamp - joinThisBeforeMs); |
||||
long timeTo = Math.max(0L, inputRecordTimestamp + joinThisAfterMs); |
||||
boolean emittedJoinWithSelf = false; |
||||
final Record selfRecord = record |
||||
.withValue(joinerThis.apply(record.key(), record.value(), (V2) record.value())) |
||||
.withTimestamp(inputRecordTimestamp); |
||||
sharedTimeTracker.advanceStreamTime(inputRecordTimestamp); |
||||
|
||||
// Join current record with other
|
||||
try (final WindowStoreIterator<V2> iter = windowStore.fetch(record.key(), timeFrom, timeTo)) { |
||||
while (iter.hasNext()) { |
||||
final KeyValue<Long, V2> otherRecord = iter.next(); |
||||
final long otherRecordTimestamp = otherRecord.key; |
||||
|
||||
// Join this with other
|
||||
context().forward( |
||||
record.withValue(joinerThis.apply( |
||||
record.key(), record.value(), otherRecord.value)) |
||||
.withTimestamp(Math.max(inputRecordTimestamp, otherRecordTimestamp))); |
||||
} |
||||
} |
||||
|
||||
// Needs to be in a different loop to ensure correct ordering of records where
|
||||
// correct ordering means it matches the output of an inner join.
|
||||
timeFrom = Math.max(0L, inputRecordTimestamp - joinOtherBeforeMs); |
||||
timeTo = Math.max(0L, inputRecordTimestamp + joinOtherAfterMs); |
||||
try (final WindowStoreIterator<V2> iter2 = windowStore.fetch(record.key(), timeFrom, timeTo)) { |
||||
while (iter2.hasNext()) { |
||||
final KeyValue<Long, V2> otherRecord = iter2.next(); |
||||
final long otherRecordTimestamp = otherRecord.key; |
||||
final long maxRecordTimestamp = Math.max(inputRecordTimestamp, otherRecordTimestamp); |
||||
|
||||
// This is needed so that output records follow timestamp order
|
||||
// Join this with self
|
||||
if (inputRecordTimestamp < maxRecordTimestamp && !emittedJoinWithSelf) { |
||||
emittedJoinWithSelf = true; |
||||
context().forward(selfRecord); |
||||
} |
||||
|
||||
// Join other with current record
|
||||
context().forward( |
||||
record |
||||
.withValue(joinerThis.apply(record.key(), (V1) otherRecord.value, (V2) record.value())) |
||||
.withTimestamp(Math.max(inputRecordTimestamp, otherRecordTimestamp))); |
||||
} |
||||
} |
||||
|
||||
// Join this with self
|
||||
if (!emittedJoinWithSelf) { |
||||
context().forward(selfRecord); |
||||
} |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,60 @@
@@ -0,0 +1,60 @@
|
||||
/* |
||||
* 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.processor.api.ProcessorContext; |
||||
import org.apache.kafka.streams.processor.api.Record; |
||||
import org.apache.kafka.streams.processor.api.RecordMetadata; |
||||
import org.slf4j.Logger; |
||||
|
||||
public final class StreamStreamJoinUtil { |
||||
|
||||
private StreamStreamJoinUtil(){ |
||||
} |
||||
|
||||
public static <KIn, VIn, KOut, VOut> boolean skipRecord( |
||||
final Record<KIn, VIn> record, final Logger logger, |
||||
final Sensor droppedRecordsSensor, |
||||
final ProcessorContext<KOut, VOut> context) { |
||||
// we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
|
||||
//
|
||||
// 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
|
||||
// furthermore, on left/outer joins 'null' in ValueJoiner#apply() indicates a missing record --
|
||||
// thus, to be consistent and to avoid ambiguous null semantics, null values are ignored
|
||||
if (record.key() == null || record.value() == null) { |
||||
if (context.recordMetadata().isPresent()) { |
||||
final RecordMetadata recordMetadata = context.recordMetadata().get(); |
||||
logger.warn( |
||||
"Skipping record due to null key or value. " |
||||
+ "topic=[{}] partition=[{}] offset=[{}]", |
||||
recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset() |
||||
); |
||||
} else { |
||||
logger.warn( |
||||
"Skipping record due to null key or value. Topic, partition, and offset not known." |
||||
); |
||||
} |
||||
droppedRecordsSensor.record(); |
||||
return true; |
||||
} else { |
||||
return false; |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,38 @@
@@ -0,0 +1,38 @@
|
||||
/* |
||||
* 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.graph; |
||||
|
||||
public class WindowedStreamProcessorNode<K, V> extends ProcessorGraphNode<K, V> { |
||||
|
||||
private final String windowStoreName; |
||||
|
||||
/** |
||||
* Create a node representing a Stream Join Window processor. |
||||
*/ |
||||
public WindowedStreamProcessorNode(final String windowStoreName, |
||||
final ProcessorParameters<K, V, ?, ?> processorParameters) { |
||||
super(processorParameters.processorName(), processorParameters); |
||||
this.windowStoreName = windowStoreName; |
||||
} |
||||
|
||||
@Override |
||||
public String toString() { |
||||
return "WindowedStreamProcessorNode{" + |
||||
"storeName=" + windowStoreName + |
||||
"} " + super.toString(); |
||||
} |
||||
} |
||||
@ -0,0 +1,337 @@
@@ -0,0 +1,337 @@
|
||||
/* |
||||
* 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 static java.time.Duration.ofMillis; |
||||
import static java.time.Duration.ofSeconds; |
||||
|
||||
import java.util.List; |
||||
import java.util.Properties; |
||||
import org.apache.kafka.common.serialization.Serdes; |
||||
import org.apache.kafka.common.serialization.StringSerializer; |
||||
import org.apache.kafka.streams.KeyValueTimestamp; |
||||
import org.apache.kafka.streams.StreamsBuilder; |
||||
import org.apache.kafka.streams.StreamsConfig; |
||||
import org.apache.kafka.streams.TestInputTopic; |
||||
import org.apache.kafka.streams.Topology; |
||||
import org.apache.kafka.streams.TopologyTestDriver; |
||||
import org.apache.kafka.streams.kstream.Consumed; |
||||
import org.apache.kafka.streams.kstream.JoinWindows; |
||||
import org.apache.kafka.streams.kstream.KStream; |
||||
import org.apache.kafka.streams.kstream.StreamJoined; |
||||
import org.apache.kafka.streams.kstream.ValueJoiner; |
||||
import org.apache.kafka.test.MockApiProcessor; |
||||
import org.apache.kafka.test.MockApiProcessorSupplier; |
||||
import org.apache.kafka.test.StreamsTestUtils; |
||||
import org.junit.Test; |
||||
|
||||
public class KStreamKStreamSelfJoinTest { |
||||
private final String topic1 = "topic1"; |
||||
private final String topic2 = "topic2"; |
||||
private final Properties props = StreamsTestUtils.getStreamsConfig(Serdes.String(), Serdes.String()); |
||||
|
||||
@Test |
||||
public void shouldMatchInnerJoinWithSelfJoinWithSingleStream() { |
||||
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST); |
||||
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE); |
||||
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2; |
||||
final List<KeyValueTimestamp<String, String>> expected; |
||||
final StreamsBuilder streamsBuilder = new StreamsBuilder(); |
||||
|
||||
// Inner join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> innerJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream2 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> innerJoin = stream2.join( |
||||
stream2, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
innerJoin.process(innerJoinSupplier); |
||||
|
||||
final Topology innerJoinTopology = streamsBuilder.build(); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(innerJoinTopology)) { |
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic2, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
innerJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 1L); |
||||
inputTopic.pipeInput("B", "1", 2L); |
||||
inputTopic.pipeInput("A", "2", 3L); |
||||
inputTopic.pipeInput("B", "2", 4L); |
||||
inputTopic.pipeInput("B", "3", 5L); |
||||
expected = processor.processed(); |
||||
} |
||||
|
||||
// Self join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> selfJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream1 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> selfJoin = stream1.join( |
||||
stream1, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
selfJoin.process(selfJoinSupplier); |
||||
|
||||
final Topology selfJoinTopology = streamsBuilder.build(props); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(selfJoinTopology, props)) { |
||||
|
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic1, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
selfJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 1L); |
||||
inputTopic.pipeInput("B", "1", 2L); |
||||
inputTopic.pipeInput("A", "2", 3L); |
||||
inputTopic.pipeInput("B", "2", 4L); |
||||
inputTopic.pipeInput("B", "3", 5L); |
||||
|
||||
// Then:
|
||||
processor.checkAndClearProcessResult(expected.toArray(new KeyValueTimestamp[0])); |
||||
} |
||||
} |
||||
|
||||
|
||||
@Test |
||||
public void shouldMatchInnerJoinWithSelfJoinWithTwoStreams() { |
||||
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST); |
||||
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE); |
||||
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2; |
||||
final List<KeyValueTimestamp<String, String>> expected; |
||||
final StreamsBuilder streamsBuilder = new StreamsBuilder(); |
||||
|
||||
// Inner join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> innerJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream3 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream4 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> innerJoin = stream3.join( |
||||
stream4, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
innerJoin.process(innerJoinSupplier); |
||||
|
||||
final Topology innerJoinTopology = streamsBuilder.build(); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(innerJoinTopology)) { |
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic2, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
innerJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 1L); |
||||
inputTopic.pipeInput("B", "1", 2L); |
||||
inputTopic.pipeInput("A", "2", 3L); |
||||
inputTopic.pipeInput("B", "2", 4L); |
||||
inputTopic.pipeInput("B", "3", 5L); |
||||
expected = processor.processed(); |
||||
} |
||||
|
||||
// Self join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> selfJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream1 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream2 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> selfJoin = stream1.join( |
||||
stream2, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofMillis(100)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
selfJoin.process(selfJoinSupplier); |
||||
|
||||
final Topology topology1 = streamsBuilder.build(props); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(topology1, props)) { |
||||
|
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic1, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
selfJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 1L); |
||||
inputTopic.pipeInput("B", "1", 2L); |
||||
inputTopic.pipeInput("A", "2", 3L); |
||||
inputTopic.pipeInput("B", "2", 4L); |
||||
inputTopic.pipeInput("B", "3", 5L); |
||||
|
||||
// Then:
|
||||
processor.checkAndClearProcessResult(expected.toArray(new KeyValueTimestamp[0])); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
public void shouldMatchInnerJoinWithSelfJoinDifferentBeforeAfterWindows() { |
||||
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST); |
||||
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE); |
||||
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2; |
||||
final List<KeyValueTimestamp<String, String>> expected; |
||||
final StreamsBuilder streamsBuilder = new StreamsBuilder(); |
||||
|
||||
// Inner join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> innerJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream3 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream4 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> innerJoin = stream3.join( |
||||
stream4, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(11), ofSeconds(10)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
innerJoin.process(innerJoinSupplier); |
||||
|
||||
final Topology innerJoinTopology = streamsBuilder.build(); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(innerJoinTopology)) { |
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic2, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
innerJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 0L); |
||||
inputTopic.pipeInput("A", "2", 11000L); |
||||
inputTopic.pipeInput("B", "1", 12000L); |
||||
inputTopic.pipeInput("A", "3", 13000L); |
||||
inputTopic.pipeInput("A", "4", 15000L); |
||||
inputTopic.pipeInput("C", "1", 16000L); |
||||
inputTopic.pipeInput("D", "1", 17000L); |
||||
inputTopic.pipeInput("A", "5", 30000L); |
||||
expected = processor.processed(); |
||||
} |
||||
|
||||
// Self join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> selfJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream1 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream2 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> selfJoin = stream1.join( |
||||
stream2, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceAndGrace(ofSeconds(11), ofSeconds(10)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
selfJoin.process(selfJoinSupplier); |
||||
final Topology selfJoinTopology = streamsBuilder.build(props); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(selfJoinTopology, props)) { |
||||
|
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic1, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
selfJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 0L); |
||||
inputTopic.pipeInput("A", "2", 11000L); |
||||
inputTopic.pipeInput("B", "1", 12000L); |
||||
inputTopic.pipeInput("A", "3", 13000L); |
||||
inputTopic.pipeInput("A", "4", 15000L); |
||||
inputTopic.pipeInput("C", "1", 16000L); |
||||
inputTopic.pipeInput("D", "1", 17000L); |
||||
inputTopic.pipeInput("A", "5", 30000L); |
||||
|
||||
// Then:
|
||||
processor.checkAndClearProcessResult(expected.toArray(new KeyValueTimestamp[0])); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
public void shouldMatchInnerJoinWithSelfJoinOutOfOrderMessages() { |
||||
props.setProperty(StreamsConfig.BUILT_IN_METRICS_VERSION_CONFIG, StreamsConfig.METRICS_LATEST); |
||||
props.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE); |
||||
final ValueJoiner<String, String, String> valueJoiner = (v, v2) -> v + v2; |
||||
final List<KeyValueTimestamp<String, String>> expected; |
||||
final StreamsBuilder streamsBuilder = new StreamsBuilder(); |
||||
|
||||
// Inner join topology
|
||||
final MockApiProcessorSupplier<String, String, Void, Void> innerJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> stream3 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream4 = streamsBuilder.stream( |
||||
topic2, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> innerJoin = stream3.join( |
||||
stream4, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofSeconds(10)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
innerJoin.process(innerJoinSupplier); |
||||
|
||||
final Topology topology2 = streamsBuilder.build(); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(topology2)) { |
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic2, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
innerJoinSupplier.theCapturedProcessor(); |
||||
|
||||
inputTopic.pipeInput("A", "1", 0L); |
||||
inputTopic.pipeInput("A", "2", 9999); |
||||
inputTopic.pipeInput("B", "1", 11000L); |
||||
inputTopic.pipeInput("A", "3", 13000L); |
||||
inputTopic.pipeInput("A", "4", 15000L); |
||||
inputTopic.pipeInput("C", "1", 16000L); |
||||
inputTopic.pipeInput("D", "1", 17000L); |
||||
inputTopic.pipeInput("A", "5", 30000L); |
||||
inputTopic.pipeInput("A", "5", 6000); |
||||
expected = processor.processed(); |
||||
} |
||||
|
||||
// Self join topology
|
||||
final KStream<String, String> stream1 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final KStream<String, String> stream2 = streamsBuilder.stream( |
||||
topic1, Consumed.with(Serdes.String(), Serdes.String())); |
||||
final MockApiProcessorSupplier<String, String, Void, Void> selfJoinSupplier = |
||||
new MockApiProcessorSupplier<>(); |
||||
final KStream<String, String> selfJoin = stream1.join( |
||||
stream2, |
||||
valueJoiner, |
||||
JoinWindows.ofTimeDifferenceWithNoGrace(ofSeconds(10)), |
||||
StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()) |
||||
); |
||||
selfJoin.process(selfJoinSupplier); |
||||
|
||||
final Topology selfJoinTopology = streamsBuilder.build(props); |
||||
try (final TopologyTestDriver driver = new TopologyTestDriver(selfJoinTopology, props)) { |
||||
|
||||
final TestInputTopic<String, String> inputTopic = |
||||
driver.createInputTopic(topic1, new StringSerializer(), new StringSerializer()); |
||||
final MockApiProcessor<String, String, Void, Void> processor = |
||||
selfJoinSupplier.theCapturedProcessor(); |
||||
inputTopic.pipeInput("A", "1", 0L); |
||||
inputTopic.pipeInput("A", "2", 9999); |
||||
inputTopic.pipeInput("B", "1", 11000L); |
||||
inputTopic.pipeInput("A", "3", 13000L); |
||||
inputTopic.pipeInput("A", "4", 15000L); |
||||
inputTopic.pipeInput("C", "1", 16000L); |
||||
inputTopic.pipeInput("D", "1", 17000L); |
||||
inputTopic.pipeInput("A", "5", 30000L); |
||||
inputTopic.pipeInput("A", "5", 6000); |
||||
|
||||
// Then:
|
||||
processor.checkAndClearProcessResult(expected.toArray(new KeyValueTimestamp[0])); |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue