A simplified Kafka Streams topology

Topology.java

KStreamBuilder builder = new KStreamBuilder();
KStream<String, Long> visitsStream = builder.stream(Serdes.String(), Serdes.Long(), "visitsTopic");
KGroupedStream<String, Long> groupedStream = visitsStream.groupByKey();
KTable<String, Long> totalCount = groupedStream.count("totalVisitCount");
KTable<Windowed<String>, Long> windowedCount = groupedStream.count(TimeWindows.of(60 * 60 * 1000), "hourlyVisitCount");
groupedStream.count(SessionWindows.with(60 * 1000), "sessionVisitCount");