From d25dc9ec4152f4d2de60abbd387c9c57e935135b Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 4 Jun 2024 23:26:38 +0200 Subject: [PATCH] splitter: 1.2.0 - Refined `SplitterApplicationConfiguration` * Separated the configuration for serialization/deserialization in a static method. --- .../SplitterApplicationConfiguration.java | 42 ++++++++++++------- 1 file changed, 28 insertions(+), 14 deletions(-) diff --git a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java index 7143e1a..92537b2 100644 --- a/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationConfiguration.java @@ -2,7 +2,6 @@ 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.context.properties.EnableConfigurationProperties; @@ -27,24 +26,13 @@ public class SplitterApplicationConfiguration @Bean(initMethod = "start", destroyMethod = "stop") public SplitterStreamProcessor streamProcessor( SplitterApplicationProperties properties, + Properties streamProcessorProperties, 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, JsonSerde.class.getName()); - propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); - propertyMap.put(JsonDeserializer.TRUSTED_PACKAGES, SplitterApplication.class.getName()); - propertyMap.put(JsonDeserializer.KEY_DEFAULT_TYPE, User.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); + streamProcessorProperties); streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) -> { @@ -59,4 +47,30 @@ public class SplitterApplicationConfiguration return streamProcessor; } + + @Bean + public Properties streamProcessorProperties( + SplitterApplicationProperties applicationProperties) + { + Properties properties = serializationConfig(); + + properties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationProperties.getApplicationId()); + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, applicationProperties.getBootstrapServer()); + properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + + return properties; + } + + static Properties serializationConfig() + { + Properties properties = new Properties(); + + properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); + properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); + properties.put(JsonDeserializer.KEY_DEFAULT_TYPE, User.class.getName()); + properties.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Recording.class.getName()); + properties.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); + + return properties; + } } -- 2.20.1