@Test public void testMerge() { String topic1 = "topic-1"; String topic2 = "topic-2"; KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> source1 = builder.stream(topic1); KStream<String, String> source2 = builder.stream(topic2); KStream<String, String> merged = builder.merge(source1, source2); MockProcessorSupplier<String, String> processorSupplier = new MockProcessorSupplier<>(); merged.process(processorSupplier); driver = new KStreamTestDriver(builder); driver.setTime(0L); driver.process(topic1, "A", "aa"); driver.process(topic2, "B", "bb"); driver.process(topic2, "C", "cc"); driver.process(topic1, "D", "dd"); assertEquals(Utils.mkList("A:aa", "B:bb", "C:cc", "D:dd"), processorSupplier.processed); }
public <K, V> KStream<K, V> merge(KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); } }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); } KStreamBuilder(); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); } KStreamBuilder(); KStream<K, V> stream(String... topics); KStream<K, V> stream(Serde<K> keySerde, Serde<V> valSerde, String... topics); KTable<K, V> table(String topic); KTable<K, V> table(Serde<K> keySerde, Serde<V> valSerde, String topic); KStream<K, V> merge(KStream<K, V>... streams); String newName(String prefix); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); } KStreamBuilder(); KStream<K, V> stream(String... topics); KStream<K, V> stream(Serde<K> keySerde, Serde<V> valSerde, String... topics); KTable<K, V> table(String topic); KTable<K, V> table(Serde<K> keySerde, Serde<V> valSerde, String topic); KStream<K, V> merge(KStream<K, V>... streams); String newName(String prefix);  }