juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java

   1 package de.juplo.kafka.wordcount.counter;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.streams.*;
   5 import org.apache.kafka.streams.kstream.Consumed;
   6 import org.apache.kafka.streams.kstream.Materialized;
   7 import org.apache.kafka.streams.kstream.Produced;
   8 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
   9 import org.apache.kafka.streams.state.QueryableStoreTypes;
  10 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  11 import org.springframework.kafka.support.serializer.JsonSerde;
  12 import org.springframework.kafka.support.serializer.JsonSerializer;
  13
  14 import java.util.Map;
  15 import java.util.Properties;
  16 import java.util.stream.Collectors;
  17
  18
  19 @Slf4j
  20 public class CounterStreamProcessor
  21 {
  22         public static final String STORE_NAME = "counter";
  23
  24
  25         public final KafkaStreams streams;
  26
  27
  28         public CounterStreamProcessor(
  29                         String inputTopic,
  30                         String outputTopic,
  31                         Properties properties,
  32                         KeyValueBytesStoreSupplier storeSupplier)
  33         {
  34                 Topology topology = CounterStreamProcessor.buildTopology(
  35                                 inputTopic,
  36                                 outputTopic,
  37                                 storeSupplier);
  38
  39                 streams = new KafkaStreams(topology, properties);
  40         }
  41
  42         static Topology buildTopology(
  43                         String inputTopic,
  44                         String outputTopic,
  45                         KeyValueBytesStoreSupplier storeSupplier)
  46         {
  47                 StreamsBuilder builder = new StreamsBuilder();
  48
  49                 builder
  50                                 .stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
  51                                 .map((key, word) -> new KeyValue<>(word, word))
  52                                 .groupByKey()
  53                                 .count(
  54                                                 Materialized
  55                                                                 .<Word, Long>as(storeSupplier)
  56                                                                 .withKeySerde(new JsonSerde<>(Word.class))) // No headers are present: fixed typing is needed!
  57                                 .toStream()
  58                                 .map((word, counter) -> new KeyValue<>(word, WordCounter.of(word, counter)))
  59                                 .to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
  60
  61                 Topology topology = builder.build();
  62                 log.info("\n\n{}", topology.describe());
  63
  64                 return topology;
  65         }
  66
  67         ReadOnlyKeyValueStore<Word, Long> getStore()
  68         {
  69                 return streams.store(StoreQueryParameters.fromNameAndType(STORE_NAME, QueryableStoreTypes.keyValueStore()));
  70         }
  71
  72         public void start()
  73         {
  74                 log.info("Starting Stream-Processor");
  75                 streams.start();
  76         }
  77
  78         public void stop()
  79         {
  80                 log.info("Stopping Stream-Processor");
  81                 streams.close();
  82         }
  83
  84
  85
  86         public static JsonSerde<User> inKeySerde()
  87         {
  88                 return new JsonSerde<>(User.class);
  89         }
  90
  91         public static JsonSerde<Word> inValueSerde()
  92         {
  93                 return new JsonSerde<>(Word.class);
  94         }
  95
  96         public static JsonSerde<Word> outKeySerde()
  97         {
  98                 return serde(true);
  99         }
 100
 101         public static JsonSerde<WordCounter> outValueSerde()
 102         {
 103                 return serde(false);
 104         }
 105
 106         public static <T> JsonSerde<T> serde(boolean isKey)
 107         {
 108                 JsonSerde<T> serde = new JsonSerde<>();
 109                 serde.configure(
 110                                 Map.of(JsonSerializer.TYPE_MAPPINGS, typeMappingsConfig()),
 111                                 isKey);
 112                 return serde;
 113         }
 114
 115         private static String typeMappingsConfig()
 116         {
 117                 return typeMappings()
 118                                 .entrySet()
 119                                 .stream()
 120                                 .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
 121                                 .collect(Collectors.joining(","));
 122         }
 123
 124         private static Map<String, Class> typeMappings()
 125         {
 126                 return Map.of(
 127                                 "word", Word.class,
 128                                 "counter", WordCounter.class);
 129         }
 130 }