WIP

[demos/kafka/wordcount] / src / main / java / de / juplo / kafka / wordcount / top10 / Top10ApplicationConfiguration.java

diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java
index cdf268f..7749917 100644 (file)
--- a/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java
@@ -1,89 +1,65 @@
 package de.juplo.kafka.wordcount.top10;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import jakarta.annotation.PostConstruct;
-import jakarta.annotation.PreDestroy;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.common.serialization.Serdes;
-import org.apache.kafka.streams.KafkaStreams;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.stereotype.Component;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+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 java.util.regex.Pattern;
 
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
 
 
+@Configuration
+@EnableConfigurationProperties(Top10ApplicationProperties.class)
 @Slf4j
-@Component
 public class Top10ApplicationConfiguration
 {
-       final static Pattern PATTERN = Pattern.compile("\\W+");
-
-       public final KafkaStreams streams;
-
-
-       public Top10ApplicationConfiguration(
-                       Top10ApplicationProperties properties,
-                       ObjectMapper mapper,
-                       ConfigurableApplicationContext context)
+       @Bean
+       public Properties streamProcessorProperties(Top10ApplicationProperties properties)
        {
-               StreamsBuilder builder = new StreamsBuilder();
-
-               builder
-                               .<String, String>stream(properties.getInputTopic())
-                               .map((keyJson, countStr) ->
-                               {
-                                       try
-                                       {
-                                               Key key = mapper.readValue(keyJson, Key.class);
-                                               Long count = Long.parseLong(countStr);
-                                               Entry entry = Entry.of(key.getWord(), count);
-                                               String entryJson = mapper.writeValueAsString(entry);
-                                               return new KeyValue<>(key.getUsername(), entryJson);
-                                       }
-                                       catch (JsonProcessingException e)
-                                       {
-                                               throw new RuntimeException(e);
-                                       }
-                               })
-                               .groupByKey()
-                               .aggregate(
-                                               () -> "{\"entries\"     : []}",
-                                               (username, entryJson, rankingJson) ->
-                                               {
-                                                       try
-                                                       {
-                                                               Ranking ranking = mapper.readValue(rankingJson, Ranking.class);
-                                                               ranking.add(mapper.readValue(entryJson, Entry.class));
-                                                               return mapper.writeValueAsString(ranking);
-                                                       }
-                                                       catch (JsonProcessingException e)
-                                                       {
-                                                               throw new RuntimeException(e);
-                                                       }
-                                               }
-                               )
-                               .toStream()
-                               .to(properties.getOutputTopic());
-
                Properties props = new Properties();
+
                props.put(StreamsConfig.APPLICATION_ID_CONFIG, properties.getApplicationId());
                props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
-               props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-               props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
+               props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
+               props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
+               props.put(JsonDeserializer.TRUSTED_PACKAGES, Top10Application.class.getPackageName());
+               props.put(JsonDeserializer.KEY_DEFAULT_TYPE, String.class.getName());
+               props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Ranking.class.getName());
+               props.put(
+                               JsonDeserializer.TYPE_MAPPINGS,
+                               "word:" + Key.class.getName() + "," +
+                               "counter:" + Entry.class.getName());
+               props.put(JsonDeserializer.REMOVE_TYPE_INFO_HEADERS, Boolean.FALSE);
+               props.put(
+                               JsonSerializer.TYPE_MAPPINGS,
+                               "ranking:" + Ranking.class.getName());
                props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 
-               streams = new KafkaStreams(builder.build(), props);
-               streams.setUncaughtExceptionHandler((Throwable e) ->
+               return props;
+       }
+
+       @Bean(initMethod = "start", destroyMethod = "stop")
+       public Top10StreamProcessor streamProcessor(
+                       Top10ApplicationProperties applicationProperties,
+                       Properties streamProcessorProperties,
+                       ConfigurableApplicationContext context)
+       {
+               Top10StreamProcessor streamProcessor = new Top10StreamProcessor(
+                               applicationProperties.getInputTopic(),
+                               applicationProperties.getOutputTopic(),
+                               streamProcessorProperties);
+
+               streamProcessor.streams.setUncaughtExceptionHandler((Throwable e) ->
                {
                        log.error("Unexpected error!", e);
                        CompletableFuture.runAsync(() ->
@@ -93,19 +69,7 @@ public class Top10ApplicationConfiguration
                        });
                        return SHUTDOWN_CLIENT;
                });
-       }
 
-       @PostConstruct
-       public void start()
-       {
-               log.info("Starting Stream-Processor");
-               streams.start();
-       }
-
-       @PreDestroy
-       public void stop()
-       {
-               log.info("Stopping Stream-Processor");
-               streams.close();
+               return streamProcessor;
        }
 }