Page History

StreamsBuilder builder = new StreamsBuilder();  

final Serde<String> stringSerde = Serdes.String();

builder  
.table(input, Consumed.with(stringSerde, stringSerde)) 
// key is not changed
.groupBy(KeyValue::pair, Grouped.with(stringSerde, stringSerde))  
.count()
.toStream()  
.to(output);

Topology topology = builder.build();