consumeAsKStream

fun <K, V> TopicDefinition<K, V>.consumeAsKStream(builder: StreamsBuilder, consumer: Consumed<K, V> = serdes.consumer("${pid}.input-stream")): KStream<K, V>