juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java

   1 package de.juplo.kafka.wordcount.top10;
   2
   3 import com.fasterxml.jackson.databind.ObjectMapper;
   4 import de.juplo.kafka.wordcount.avro.Entry;
   5 import de.juplo.kafka.wordcount.avro.Key;
   6 import de.juplo.kafka.wordcount.avro.Ranking;
   7 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
   8 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
   9 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerializer;
  10 import lombok.extern.slf4j.Slf4j;
  11 import org.apache.kafka.clients.consumer.ConsumerConfig;
  12 import org.apache.kafka.common.serialization.Serdes;
  13 import org.apache.kafka.streams.*;
  14 import org.apache.kafka.streams.kstream.Consumed;
  15 import org.apache.kafka.streams.kstream.Grouped;
  16 import org.apache.kafka.streams.kstream.Materialized;
  17 import org.apache.kafka.streams.kstream.Produced;
  18 import org.springframework.boot.SpringApplication;
  19 import org.springframework.context.ConfigurableApplicationContext;
  20 import org.springframework.stereotype.Component;
  21
  22 import javax.annotation.PostConstruct;
  23 import javax.annotation.PreDestroy;
  24 import java.util.Arrays;
  25 import java.util.LinkedList;
  26 import java.util.List;
  27 import java.util.Properties;
  28 import java.util.concurrent.CompletableFuture;
  29 import java.util.regex.Pattern;
  30
  31 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
  32
  33
  34 @Slf4j
  35 @Component
  36 public class Top10StreamProcessor
  37 {
  38         final static Pattern PATTERN = Pattern.compile("\\W+");
  39
  40         public final KafkaStreams streams;
  41
  42
  43         public Top10StreamProcessor(
  44                         Top10ApplicationProperties properties,
  45                         ObjectMapper mapper,
  46                         ConfigurableApplicationContext context)
  47         {
  48                 StreamsBuilder builder = new StreamsBuilder();
  49
  50                 builder
  51                                 .<Key, Long>stream(properties.getInputTopic(), Consumed.with(null, Serdes.Long()))
  52                                 .map((key, count) -> new KeyValue<>(
  53                                                 key.getUsername(),
  54                                                 Entry.newBuilder().setWord(key.getWord()).setCount(count).build()))
  55                                 .groupByKey(Grouped.keySerde(Serdes.String()))
  56                                 .aggregate(
  57                                                 () -> Ranking.newBuilder().setEntries(new LinkedList<Entry>()).build(),
  58                                                 (username, newEntry, ranking) -> {
  59                                                         List<Entry> entries = new LinkedList<>(ranking.getEntries());
  60
  61                                                         if (entries.isEmpty())
  62                                                         {
  63                                                                 entries.add(newEntry);
  64                                                         }
  65                                                         else
  66                                                         {
  67                                                                 for (int i = 0; i < entries.size(); i++)
  68                                                                 {
  69                                                                         Entry entry = entries.get(i);
  70                                                                         if (entry.getCount() <= newEntry.getCount())
  71                                                                         {
  72                                                                                 entries.add(i, newEntry);
  73                                                                                 for (int j = i + 1; j < entries.size(); j++)
  74                                                                                 {
  75                                                                                         entry = entries.get(j);
  76                                                                                         if (entry.getWord().equals(newEntry.getWord()))
  77                                                                                         {
  78                                                                                                 entries.remove(j);
  79                                                                                                 break;
  80                                                                                         }
  81                                                                                 }
  82                                                                                 if (entries.size() > 10)
  83                                                                                 {
  84                                                                                         entries = entries.subList(0, 10);
  85                                                                                 }
  86                                                                         }
  87                                                                 }
  88                                                         }
  89
  90                                                         ranking.setEntries(entries);
  91                                                         return ranking;
  92                                                 })
  93                                 .toStream()
  94                                 .to(properties.getOutputTopic(), Produced.keySerde(Serdes.String()));
  95
  96                 Properties props = new Properties();
  97                 props.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId());
  98                 props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
  99                 props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
 100                 props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
 101                 props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, properties.getSchemaRegistry());
 102                 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 103
 104                 Topology topology = builder.build();
 105                 log.info("Topology:\n-----------------\n\n{}-----------------", topology.describe());
 106
 107                 streams = new KafkaStreams(topology, props);
 108                 streams.setUncaughtExceptionHandler((Throwable e) ->
 109                 {
 110                         log.error("Unexpected error!", e);
 111                         CompletableFuture.runAsync(() ->
 112                         {
 113                                 log.info("Stopping application...");
 114                                 SpringApplication.exit(context, () -> 1);
 115                         });
 116                         return SHUTDOWN_CLIENT;
 117                 });
 118         }
 119
 120         @PostConstruct
 121         public void start()
 122         {
 123                 log.info("Starting Stream-Processor");
 124                 streams.start();
 125         }
 126
 127         @PreDestroy
 128         public void stop()
 129         {
 130                 log.info("Stopping Stream-Processor");
 131                 streams.close();
 132         }
 133 }