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;
8 import java.util.Properties;
12 public class Top10StreamProcessor
14 public final KafkaStreams streams;
17 public Top10StreamProcessor(
22 StreamsBuilder builder = new StreamsBuilder();
25 .<Key, Counter>stream(inputTopic)
26 .map((key, counter) ->
28 Entry entry = Entry.of(key.getWord(), counter.getCounter());
29 return new KeyValue<>(key.getUser(), entry);
34 (user, entry, ranking) ->
43 streams = new KafkaStreams(builder.build(), props);
48 log.info("Starting Stream-Processor");
54 log.info("Stopping Stream-Processor");