juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/popular/PopularApplicationConfiguriation.java

   1 package de.juplo.kafka.wordcount.popular;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.ConsumerConfig;
   5 import org.apache.kafka.streams.StreamsConfig;
   6 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
   7 import org.apache.kafka.streams.state.Stores;
   8 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
   9 import org.springframework.boot.SpringApplication;
  10 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  11 import org.springframework.context.ConfigurableApplicationContext;
  12 import org.springframework.context.annotation.Bean;
  13 import org.springframework.context.annotation.Configuration;
  14 import org.springframework.kafka.support.serializer.JsonDeserializer;
  15 import org.springframework.kafka.support.serializer.JsonSerde;
  16
  17 import java.time.Duration;
  18 import java.time.ZoneId;
  19 import java.util.Properties;
  20 import java.util.concurrent.CompletableFuture;
  21
  22 import static de.juplo.kafka.wordcount.popular.PopularStreamProcessor.KEY_VALUE_STORE_NAME;
  23 import static de.juplo.kafka.wordcount.popular.PopularStreamProcessor.WINDOW_STORE_NAME;
  24 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
  25
  26
  27 @Configuration
  28 @EnableConfigurationProperties(PopularApplicationProperties.class)
  29 @Slf4j
  30 public class PopularApplicationConfiguriation
  31 {
  32         @Bean
  33         public Properties streamProcessorProperties(
  34                         PopularApplicationProperties counterProperties)
  35         {
  36                 Properties propertyMap = serializationConfig();
  37
  38                 propertyMap.put(StreamsConfig.APPLICATION_ID_CONFIG, counterProperties.getApplicationId());
  39
  40                 propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, counterProperties.getBootstrapServer());
  41                 if (counterProperties.getCommitInterval() != null)
  42                         propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, counterProperties.getCommitInterval());
  43                 if (counterProperties.getCacheMaxBytes() != null)
  44                         propertyMap.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, counterProperties.getCacheMaxBytes());
  45
  46                 propertyMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  47
  48                 return propertyMap;
  49         }
  50
  51         static Properties serializationConfig()
  52         {
  53                 Properties propertyMap = new Properties();
  54
  55                 propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
  56                 propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
  57                 propertyMap.put(JsonDeserializer.TRUSTED_PACKAGES, PopularApplication.class.getPackageName());
  58
  59                 return propertyMap;
  60         }
  61
  62         @Bean(initMethod = "start", destroyMethod = "stop")
  63         public PopularStreamProcessor streamProcessor(
  64                         PopularApplicationProperties applicationProperties,
  65                         Properties streamProcessorProperties,
  66                         ZoneId zone,
  67                         KeyValueBytesStoreSupplier keyValueBytesStoreSupplier,
  68                         WindowBytesStoreSupplier windowBytesStoreSupplier,
  69                         ConfigurableApplicationContext context)
  70         {
  71                 PopularStreamProcessor streamProcessor = new PopularStreamProcessor(
  72                                 applicationProperties.getInputTopic(),
  73                                 applicationProperties.getOutputTopic(),
  74                                 streamProcessorProperties,
  75                                 zone,
  76                                 windowBytesStoreSupplier,
  77                                 keyValueBytesStoreSupplier);
  78
  79                 streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
  80                 {
  81                         log.error("Unexpected error!", e);
  82                         CompletableFuture.runAsync(() ->
  83                         {
  84                                 log.info("Stopping application...");
  85                                 SpringApplication.exit(context, () -> 1);
  86                         });
  87                         return SHUTDOWN_CLIENT;
  88                 });
  89
  90
  91                 return streamProcessor;
  92         }
  93
  94         @Bean
  95         public ZoneId defaultZone()
  96         {
  97                 return ZoneId.systemDefault();
  98         }
  99
 100         @Bean
 101         public WindowBytesStoreSupplier windowBytesStoreSupplier()
 102         {
 103                 return Stores.persistentWindowStore(
 104                                 KEY_VALUE_STORE_NAME,
 105                                 Duration.ofSeconds(60),
 106                                 Duration.ofSeconds(30),
 107                                 true);
 108         }
 109
 110         @Bean
 111         public KeyValueBytesStoreSupplier keyValueBytesStoreSupplier()
 112         {
 113                 return Stores.persistentKeyValueStore(WINDOW_STORE_NAME);
 114         }
 115 }