From 9c784dad94b97334cc55d90326f4810b7aa9a199 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 26 May 2024 11:34:51 +0200 Subject: [PATCH] splitter: 1.1.6 - Moved config `SplitterApplicationConfiguration` -- COPY --- .../SplitterApplicationConfiguration.java | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java diff --git a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java new file mode 100644 index 0000000..6be4b59 --- /dev/null +++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java @@ -0,0 +1,67 @@ +package de.juplo.kafka.wordcount.splitter; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.streams.StreamsConfig; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.support.serializer.JsonSerializer; + +import java.util.Properties; +import java.util.concurrent.CompletableFuture; + +import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT; + + +@SpringBootApplication +@EnableConfigurationProperties(SplitterApplicationProperties.class) +@Slf4j +public class SplitterApplicationConfiguration +{ + @Bean(initMethod = "start", destroyMethod = "stop") + public SplitterStreamProcessor streamProcessor( + SplitterApplicationProperties properties, + ConfigurableApplicationContext context) + { + Properties propertyMap = new Properties(); + + propertyMap.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId()); + propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer()); + propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName()); + propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); + propertyMap.put(JsonDeserializer.TRUSTED_PACKAGES, Recording.class.getName() ); + propertyMap.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Recording.class.getName()); + propertyMap.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); + propertyMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + + SplitterStreamProcessor streamProcessor = new SplitterStreamProcessor( + properties.getInputTopic(), + properties.getOutputTopic(), + propertyMap); + + streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) -> + { + log.error("Unexpected error!", e); + CompletableFuture.runAsync(() -> + { + log.info("Stopping application..."); + SpringApplication.exit(context, () -> 1); + }); + return SHUTDOWN_CLIENT; + }); + + + return streamProcessor; + } + + public static void main(String[] args) + { + SpringApplication.run(SplitterApplicationConfiguration.class, args); + } +} -- 2.20.1