Der Consumer kann mit mehreren Topics konfiguriert werden

diff --git a/README.sh b/README.sh
index b32d9f5ef54d5bfbc539e553d177a97875e3c430..5e265f920305f11118b3a050e99edda7740e2ef5 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/spring-consumer:1.0-SNAPSHOT
+IMAGE=juplo/spring-consumer:1.1-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index 715dd8e582cd15c1188814fb71e3890aa4c031b1..a533c73b2c88c86a24f2fdb76a1619b5c7e6ea56 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -216,7 +216,7 @@ services:
     command: kafka:9092 test producer
 
   consumer:
-    image: juplo/spring-consumer:1.0-SNAPSHOT
+    image: juplo/spring-consumer:1.1-SNAPSHOT
     environment:
       spring.kafka.bootstrap-servers: kafka:9092
       spring.kafka.client-id: consumer

diff --git a/pom.xml b/pom.xml
index 13ded014b22d69b46eb2370ca999a391f8238fa3..f161e17b507466c8063dd04cff35820bff90def0 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -15,7 +15,7 @@
   <artifactId>spring-consumer</artifactId>
   <name>Spring Consumer</name>
   <description>Super Simple Consumer-Group, that is implemented as Spring-Boot application and configured by Spring Kafka</description>
-  <version>1.0-SNAPSHOT</version>
+  <version>1.1-SNAPSHOT</version>
 
   <properties>
     <java.version>17</java.version>

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 62d61a20905dd8c4e31089199f38bc11a5b92332..9a237b77e24a4b58b0bebc9faf8c9de9caac3264 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -22,7 +22,7 @@ public class ApplicationConfiguration
     return
         new SimpleConsumer(
             kafkaProperties.getClientId(),
-            applicationProperties.getTopic(),
+            applicationProperties.getTopics(),
             kafkaConsumer);
   }
 

diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java
index a4cc8b89b98cc7225633edbb53c6aabe2041e61e..4092fa2634840f641da3d4cb57fd1bf46e5cc8fd 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationProperties.java
+++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java
@@ -17,5 +17,5 @@ public class ApplicationProperties
 {
   @NotNull
   @NotEmpty
-  private String topic;
+  private String[] topics;
 }

diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java
index aadc11f61be58a4057f1350cee3384e45ffb0d52..a76132b17c278609d5c05dec472b909d17345e45 100644 (file)
--- a/src/main/java/de/juplo/kafka/SimpleConsumer.java
+++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java
@@ -17,7 +17,7 @@ import java.util.concurrent.Callable;
 public class SimpleConsumer implements Callable<Integer>
 {
   private final String id;
-  private final String topic;
+  private final String[] topics;
   private final Consumer<String, String> consumer;
 
   private long consumed = 0;
@@ -28,8 +28,8 @@ public class SimpleConsumer implements Callable<Integer>
   {
     try
     {
-      log.info("{} - Subscribing to topic {}", id, topic);
-      consumer.subscribe(Arrays.asList(topic));
+      log.info("{} - Subscribing to topics: {}", id, topics);
+      consumer.subscribe(Arrays.asList(topics));
 
       while (true)
       {

diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index b7fedad35f872d23da5f22b2f942bc91448b6b7d..50b114da6e7bfddfeed9ff23f08d5b2cf9e6145d 100644 (file)
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -1,6 +1,6 @@
 simple:
   consumer:
-    topic: test
+    topics: test
 management:
   endpoint:
     shutdown:
@@ -19,7 +19,7 @@ info:
     bootstrap-server: ${spring.kafka.bootstrap-servers}
     client-id: ${spring.kafka.client-id}
     group-id: ${spring.kafka.consumer.group-id}
-    topic: ${simple.consumer.topic}
+    topics: ${simple.consumer.topics}
     auto-offset-reset: ${spring.kafka.consumer.auto-offset-reset}
 spring:
   kafka:

diff --git a/src/test/java/de/juplo/kafka/ApplicationIT.java b/src/test/java/de/juplo/kafka/ApplicationIT.java
index 1baca9990bea01eab50ecf4fd77296c8d5e193f1..584773ed5e5b2893623ae48e26e654d9ea444eea 100644 (file)
--- a/src/test/java/de/juplo/kafka/ApplicationIT.java
+++ b/src/test/java/de/juplo/kafka/ApplicationIT.java
@@ -14,7 +14,7 @@ import static de.juplo.kafka.ApplicationIT.TOPIC;
     webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
     properties = {
         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
-        "simple.consumer.topic=" + TOPIC })
+        "simple.consumer.topics=" + TOPIC })
 @EmbeddedKafka(topics = TOPIC)
 public class ApplicationIT
 {