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, Entry>stream(inputTopic)
38 .map((key, entry) -> new KeyValue<>(User.of(key.getUser()), entry))
42 (user, entry, ranking) -> ranking.add(entry))
46 Topology topology = builder.build();
47 log.info("\n\n{}", topology.describe());
54 log.info("Starting Stream-Processor");
60 log.info("Stopping Stream-Processor");