cleanup

module/app/data-generator/src/main/resources/application.yaml

@@ -47,5 +47,4 @@ app:
   messaging:
     topics:
       SENSOR_DATA:
-        name: sensor-data
-        partition-key: partitionKey
+        name: sensor-data

module/app/listener/src/main/resources/application.yaml

@@ -4,7 +4,7 @@ spring:
   kafka:
     bootstrap-servers: localhost:9092
     consumer:
-      group-id: group_id
+      group-id: listener
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
       properties:

module/lib/messaging-starter/src/main/java/com/maemresen/k8s/workshop/messaging/starter/MessagingAutoConfiguration.java

@@ -1,17 +1,35 @@
 package com.maemresen.k8s.workshop.messaging.starter;
 
+import org.apache.kafka.clients.admin.NewTopic;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.KafkaAdmin;
 
 @Configuration
 public class MessagingAutoConfiguration {
 
-  @ConfigurationProperties(prefix = "app.messaging")
-  @ConditionalOnMissingBean
-  @Bean
-  public MessagingProps messagingProps() {
-    return new MessagingProps();
-  }
+    @ConfigurationProperties(prefix = "app.messaging")
+    @ConditionalOnMissingBean
+    @Bean
+    public MessagingProps messagingProps() {
+        return new MessagingProps();
+    }
+
+    @Bean
+    public KafkaAdmin.NewTopics topics() {
+        return new KafkaAdmin.NewTopics(messagingProps().getTopics()
+                .values()
+                .stream()
+                .map(this::newTopic)
+                .toArray(NewTopic[]::new));
+
+    }
+
+    private NewTopic newTopic(final MessagingProps.TopicProps topicProps) {
+        return TopicBuilder.name(topicProps.name())
+                .build();
+    }
 }