BasicStream.java

StreamsBuilder builder = new StreamsBuilder();
builder
    .stream("orders")
    .flatMap(Ops::splitOrders)
    .to("order-items");
Topology topology = builder.build();
 
Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "mySampleApp");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 8);
 
KafkaStreams streams = new KafkaStreams(topology, config);
streams.start();