From 629f04d7a6d252bae7d2d8fd7e25c86ba14c0e2a Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Fri, 1 Dec 2023 17:37:51 +0100 Subject: [PATCH] Der Consumer kann mit mehreren Topics konfiguriert werden --- README.sh | 2 +- docker/docker-compose.yml | 1 + pom.xml | 2 +- src/main/java/de/juplo/kafka/ApplicationConfiguration.java | 2 +- src/main/java/de/juplo/kafka/ApplicationProperties.java | 2 +- src/main/java/de/juplo/kafka/SimpleConsumer.java | 6 +++--- src/main/resources/application.yml | 2 +- 7 files changed, 9 insertions(+), 8 deletions(-) diff --git a/README.sh b/README.sh index b32d9f5..5e265f9 100755 --- 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 715dd8e..e4191ec 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -218,6 +218,7 @@ services: consumer: image: juplo/spring-consumer:1.0-SNAPSHOT environment: + simple.consumer.topics: test spring.kafka.bootstrap-servers: kafka:9092 spring.kafka.client-id: consumer diff --git a/pom.xml b/pom.xml index d5a28ea..f688913 100644 --- a/pom.xml +++ b/pom.xml @@ -15,7 +15,7 @@ spring-consumer Spring Consumer Super Simple Consumer-Group, that is implemented as Spring-Boot application and configured by Spring Kafka - 1.0-SNAPSHOT + 1.1-SNAPSHOT 17 diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java index 62d61a2..9a237b7 100644 --- 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 a4cc8b8..4092fa2 100644 --- 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 aadc11f..a76132b 100644 --- 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 { private final String id; - private final String topic; + private final String[] topics; private final Consumer consumer; private long consumed = 0; @@ -28,8 +28,8 @@ public class SimpleConsumer implements Callable { 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 b7fedad..5143336 100644 --- 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: -- 2.20.1