@Test public void shouldNotTryProcessingFromSinkTopic() { final KStream<String, String> source = builder.stream("topic-source"); source.to("topic-sink"); final MockProcessorSupplier<String, String> processorSupplier = new MockProcessorSupplier<>(); source.process(processorSupplier); driver = new KStreamTestDriver(builder); driver.setTime(0L); driver.process("topic-source", "A", "aa"); assertEquals(Utils.mkList("A:aa"), processorSupplier.processed); }
public <K, V> KStream<K, V> stream(final String... topics) { return stream(null, null, null, null, topics); }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> stream(final String... topics) { return stream(null, null, null, null, topics); } }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> stream(final String... topics) { return stream(null, null, null, null, topics); }  }
KStreamBuilder extends TopologyBuilder { public <K, V> KStream<K, V> stream(final String... topics) { return stream(null, null, null, null, topics); }  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> stream(final String... topics) { return stream(null, null, null, null, topics); }  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);  }