juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/splitter/SplitterStreamProcessor.java

   1 package de.juplo.kafka.wordcount.splitter;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.ConsumerConfig;
   5 import org.apache.kafka.common.serialization.Serdes;
   6 import org.apache.kafka.streams.KafkaStreams;
   7 import org.apache.kafka.streams.StreamsBuilder;
   8 import org.apache.kafka.streams.StreamsConfig;
   9 import org.apache.kafka.streams.kstream.KStream;
  10 import org.springframework.boot.SpringApplication;
  11 import org.springframework.context.ConfigurableApplicationContext;
  12 import org.springframework.stereotype.Component;
  13
  14 import javax.annotation.PostConstruct;
  15 import javax.annotation.PreDestroy;
  16 import java.util.Arrays;
  17 import java.util.Properties;
  18 import java.util.concurrent.CompletableFuture;
  19 import java.util.regex.Pattern;
  20
  21 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
  22
  23
  24 @Slf4j
  25 @Component
  26 public class SplitterStreamProcessor
  27 {
  28         final static Pattern PATTERN = Pattern.compile("\\W+");
  29
  30         public final KafkaStreams streams;
  31
  32
  33         public SplitterStreamProcessor(
  34                         SplitterApplicationProperties properties,
  35                         ConfigurableApplicationContext context)
  36         {
  37                 StreamsBuilder builder = new StreamsBuilder();
  38
  39                 KStream<String, String> source = builder.stream(properties.getInputTopic());
  40                 source
  41                                 .flatMapValues(sentence -> Arrays.asList(PATTERN.split(sentence)))
  42                                 .to(properties.getOutputTopic());
  43
  44                 Properties props = new Properties();
  45                 props.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId());
  46                 props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
  47                 props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  48                 props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
  49                 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  50
  51                 streams = new KafkaStreams(builder.build(), props);
  52                 streams.setUncaughtExceptionHandler((Throwable e) ->
  53                 {
  54                         log.error("Unexpected error!", e);
  55                         CompletableFuture.runAsync(() ->
  56                         {
  57                                 log.info("Stopping application...");
  58                                 SpringApplication.exit(context, () -> 1);
  59                         });
  60                         return SHUTDOWN_CLIENT;
  61                 });
  62         }
  63
  64         @PostConstruct
  65         public void start()
  66         {
  67                 log.info("Starting Stream-Processor");
  68                 streams.start();
  69         }
  70
  71         @PreDestroy
  72         public void stop()
  73         {
  74                 log.info("Stopping Stream-Processor");
  75                 streams.close();
  76         }
  77 }