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;
8 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
9 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
11 import java.util.Properties;
15 public class Top10StreamProcessor
17 public final KafkaStreams streams;
20 public Top10StreamProcessor(
25 Topology topology = Top10StreamProcessor.buildTopology(
30 streams = new KafkaStreams(topology, props);
33 static Topology buildTopology(
36 KeyValueBytesStoreSupplier storeSupplier)
38 StreamsBuilder builder = new StreamsBuilder();
41 .<Key, Entry>stream(inputTopic)
42 .map((key, entry) -> new KeyValue<>(User.of(key.getUser()), entry))
46 (user, entry, ranking) -> ranking.add(entry))
50 Topology topology = builder.build();
51 log.info("\n\n{}", topology.describe());
56 ReadOnlyKeyValueStore<User, Ranking> getStore(String name)
63 log.info("Starting Stream-Processor");
69 log.info("Stopping Stream-Processor");