@Test public void testMerge() { final String topic1 = "topic-1"; final String topic2 = "topic-2"; final KStream<String, String> source1 = builder.stream(topic1); final KStream<String, String> source2 = builder.stream(topic2); final KStream<String, String> merged = builder.merge(source1, source2); final 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(final KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(final KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); } }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(final KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); }  }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(final KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); }  KStream<K, V> stream(final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final String... topics); KStream<K, V> stream(final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset, final Pattern topicPattern); KStream<K, V> stream(final Serde<K> keySerde, final Serde<V> valSerde, final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KTable<K, V> table(final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic); KTable<K, V> table(final TimestampExtractor timestampExtractor,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); GlobalKTable<K, V> globalTable(final String topic,
                                                 final String queryableStoreName); GlobalKTable<K, V> globalTable(final String topic); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final TimestampExtractor timestampExtractor,
                                                 final String topic,
                                                 final String queryableStoreName); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic,
                                                 final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic,
                                                 final String queryableStoreName); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic); KStream<K, V> merge(final KStream<K, V>... streams); String newName(final String prefix); String newStoreName(final String prefix); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> merge(final KStream<K, V>... streams) { return KStreamImpl.merge(this, streams); }  KStream<K, V> stream(final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final String... topics); KStream<K, V> stream(final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset, final Pattern topicPattern); KStream<K, V> stream(final Serde<K> keySerde, final Serde<V> valSerde, final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final String... topics); KStream<K, V> stream(final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KStream<K, V> stream(final AutoOffsetReset offsetReset,
                                       final TimestampExtractor timestampExtractor,
                                       final Serde<K> keySerde,
                                       final Serde<V> valSerde,
                                       final Pattern topicPattern); KTable<K, V> table(final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final String topic); KTable<K, V> table(final TimestampExtractor timestampExtractor,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); KTable<K, V> table(final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String storeName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final String queryableStoreName); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic); KTable<K, V> table(final AutoOffsetReset offsetReset,
                                     final TimestampExtractor timestampExtractor,
                                     final Serde<K> keySerde,
                                     final Serde<V> valSerde,
                                     final String topic,
                                     final StateStoreSupplier<KeyValueStore> storeSupplier); GlobalKTable<K, V> globalTable(final String topic,
                                                 final String queryableStoreName); GlobalKTable<K, V> globalTable(final String topic); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final TimestampExtractor timestampExtractor,
                                                 final String topic,
                                                 final String queryableStoreName); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic,
                                                 final StateStoreSupplier<KeyValueStore> storeSupplier); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic,
                                                 final String queryableStoreName); @SuppressWarnings("unchecked") GlobalKTable<K, V> globalTable(final Serde<K> keySerde,
                                                 final Serde<V> valSerde,
                                                 final String topic); KStream<K, V> merge(final KStream<K, V>... streams); String newName(final String prefix); String newStoreName(final String prefix);  }