1 package de.juplo.kafka.wordcount.top10;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.streams.KafkaStreams;
5 import org.apache.kafka.streams.KeyValue;
6 import org.apache.kafka.streams.StreamsBuilder;
7 import org.apache.kafka.streams.Topology;
9 import java.util.Properties;
13 public class Top10StreamProcessor
15 public final KafkaStreams streams;
18 public Top10StreamProcessor(
23 Topology topology = Top10StreamProcessor.buildTopology(
27 streams = new KafkaStreams(topology, props);
30 static Topology buildTopology(
34 StreamsBuilder builder = new StreamsBuilder();
37 .<Key, Counter>stream(inputTopic)
38 .map((key, counter) ->
40 Entry entry = Entry.of(key.getWord(), counter.getCounter());
41 return new KeyValue<>(key.getUser(), entry);
46 (user, entry, ranking) ->
55 Topology topology = builder.build();
56 log.info("\n\n{}", topology.describe());
63 log.info("Starting Stream-Processor");
69 log.info("Stopping Stream-Processor");