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.*;
  23 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
  24
  25
  26 @Configuration
  27 @EnableConfigurationProperties(PopularApplicationProperties.class)
  28 @Slf4j
  29 public class PopularApplicationConfiguriation
  30 {
  31         @Bean
  32         public Properties streamProcessorProperties(
  33                         PopularApplicationProperties counterProperties)
  34         {
  35                 Properties propertyMap = serializationConfig();
  36
  37                 propertyMap.put(StreamsConfig.APPLICATION_ID_CONFIG, counterProperties.getApplicationId());
  38
  39                 propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, counterProperties.getBootstrapServer());
  40                 if (counterProperties.getCommitInterval() != null)
  41                         propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, counterProperties.getCommitInterval());
  42                 if (counterProperties.getCacheMaxBytes() != null)
  43                         propertyMap.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, counterProperties.getCacheMaxBytes());
  44
  45                 propertyMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  46
  47                 return propertyMap;
  48         }
  49
  50         static Properties serializationConfig()
  51         {
  52                 Properties propertyMap = new Properties();
  53
  54                 propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
  55                 propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
  56                 propertyMap.put(JsonDeserializer.TRUSTED_PACKAGES, PopularApplication.class.getPackageName());
  57
  58                 return propertyMap;
  59         }
  60
  61         @Bean(initMethod = "start", destroyMethod = "stop")
  62         public PopularStreamProcessor streamProcessor(
  63                         PopularApplicationProperties applicationProperties,
  64                         Properties streamProcessorProperties,
  65                         ZoneId zone,
  66                         KeyValueBytesStoreSupplier keyValueBytesStoreSupplier,
  67                         WindowBytesStoreSupplier windowBytesStoreSupplier,
  68                         ConfigurableApplicationContext context)
  69         {
  70                 PopularStreamProcessor streamProcessor = new PopularStreamProcessor(
  71                                 applicationProperties.getInputTopic(),
  72                                 applicationProperties.getOutputTopic(),
  73                                 streamProcessorProperties,
  74                                 zone,
  75                                 windowBytesStoreSupplier,
  76                                 keyValueBytesStoreSupplier);
  77
  78                 streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
  79                 {
  80                         log.error("Unexpected error!", e);
  81                         CompletableFuture.runAsync(() ->
  82                         {
  83                                 log.info("Stopping application...");
  84                                 SpringApplication.exit(context, () -> 1);
  85                         });
  86                         return SHUTDOWN_CLIENT;
  87                 });
  88
  89
  90                 return streamProcessor;
  91         }
  92
  93         @Bean
  94         public ZoneId defaultZone()
  95         {
  96                 return ZoneId.systemDefault();
  97         }
  98
  99         @Bean
 100         public WindowBytesStoreSupplier windowBytesStoreSupplier()
 101         {
 102                 return Stores.persistentWindowStore(
 103                                 KEY_VALUE_STORE_NAME,
 104                                 WINDOW_SIZE.multipliedBy(2),
 105                                 WINDOW_SIZE,
 106                                 false); // << Must always be `false` for normal use-cases!
 107         }
 108
 109         @Bean
 110         public KeyValueBytesStoreSupplier keyValueBytesStoreSupplier()
 111         {
 112                 return Stores.persistentKeyValueStore(WINDOW_STORE_NAME);
 113         }
 114 }