// Scala API for Kafka Streams
import org.apache.kafka.streams.scala._
import ImplicitConversions._
import Serdes._

val builder = new StreamsBuilder
builder
  .stream[String, String]("input")
  .peek { (k,v) => println(s"($k, $v)") }
  .foreach { (k,v) => println(s"($k, $v)") }
val topology = builder.build
scala> println(topology.describe)
Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input])
      --> KSTREAM-PEEK-0000000001
    Processor: KSTREAM-PEEK-0000000001 (stores: [])
      --> KSTREAM-FOREACH-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-FOREACH-0000000002 (stores: [])
      --> none
      <-- KSTREAM-PEEK-0000000001