Handling der Nachricht in das Interface `RecordHandler` verlegt

diff --git a/README.sh b/README.sh
index a7cbb894146ec0e9a6bb40a9d011ba7661cdf748..203c22bd9ebaa9fee39be1031accfd594f5ee977 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/spring-consumer:1.1-generics-SNAPSHOT
+IMAGE=juplo/spring-consumer:1.1-record-handler-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then

diff --git a/build.gradle b/build.gradle
index 7e70329c0e5a86eae9f94781443938dbccd12270..f700918b9a0108030707fa941c2471cc940c129a 100644 (file)
--- a/build.gradle
+++ b/build.gradle
@@ -8,7 +8,7 @@ plugins {
 }
 
 group = 'de.juplo.kafka'
-version = '1.1-generics-SNAPSHOT'
+version = '1.1-record-handler-SNAPSHOT'
 
 java {
        toolchain {

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index e5b9a9fe0e48427905e86d6cae00320117dbc656..bf306f686f9810bc29761932c8b72da839637b3c 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -140,7 +140,7 @@ services:
     command: kafka:9092 test producer
 
   consumer:
-    image: juplo/spring-consumer:1.1-generics-SNAPSHOT
+    image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
     environment:
       spring.kafka.bootstrap-servers: kafka:9092
       spring.kafka.client-id: consumer
@@ -149,7 +149,7 @@ services:
       juplo.consumer.topic: test
 
   peter:
-    image: juplo/spring-consumer:1.1-generics-SNAPSHOT
+    image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
     environment:
       spring.kafka.bootstrap-servers: kafka:9092
       spring.kafka.client-id: consumer
@@ -158,7 +158,7 @@ services:
       juplo.consumer.topic: test
 
   ute:
-    image: juplo/spring-consumer:1.1-generics-SNAPSHOT
+    image: juplo/spring-consumer:1.1-record-handler-SNAPSHOT
     environment:
       spring.kafka.bootstrap-servers: kafka:9092
       spring.kafka.client-id: consumer

diff --git a/pom.xml b/pom.xml
index 250f40d9f281c0183cea837031b3725c9111caf3..7590d25a6e784353ae7ec9e7f595722803c9016e 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.1-generics-SNAPSHOT</version>
+  <version>1.1-record-handler-SNAPSHOT</version>
 
   <properties>
     <java.version>21</java.version>

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index e523a20327c32dbdf1404e47513d4e8318fe156c..87331b34b0b20afb64207b9ef78039aaa9b96331 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -1,5 +1,6 @@
 package de.juplo.kafka;
 
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -11,11 +12,13 @@ import org.springframework.kafka.core.ConsumerFactory;
 
 @Configuration
 @EnableConfigurationProperties(ApplicationProperties.class)
+@Slf4j
 public class ApplicationConfiguration
 {
   @Bean
   public ExampleConsumer<String, String> exampleConsumer(
     Consumer<String, String> kafkaConsumer,
+    RecordHandler<String, String> recordHandler,
     ApplicationProperties properties,
     KafkaProperties kafkaProperties,
     ConfigurableApplicationContext applicationContext)
@@ -25,9 +28,16 @@ public class ApplicationConfiguration
         kafkaProperties.getClientId(),
         properties.getConsumerProperties().getTopic(),
         kafkaConsumer,
+        recordHandler,
         () -> applicationContext.close());
   }
 
+  @Bean
+  public RecordHandler<String, String> recordHandler()
+  {
+    return (topic, partition, offset, key, value) -> log.info("No-Ops Handler called for {}={}", key, value);
+  }
+
   @Bean(destroyMethod = "")
   public Consumer<?, ?> kafkaConsumer(ConsumerFactory<?, ?> consumerFactory)
   {

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 101abd1c0b37e0c8bef761ebbfa44df71319cc94..e5a8b3d53b77594cafb1ec80c2463649692e6a78 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -16,6 +16,7 @@ public class ExampleConsumer<K, V> implements Runnable
   private final String id;
   private final String topic;
   private final Consumer<K, V> consumer;
+  private final RecordHandler<K, V> recordHandler;
   private final Thread workerThread;
   private final Runnable closeCallback;
 
@@ -26,11 +27,13 @@ public class ExampleConsumer<K, V> implements Runnable
     String clientId,
     String topic,
     Consumer<K, V> consumer,
+    RecordHandler<K, V> recordHandler,
     Runnable closeCallback)
   {
     this.id = clientId;
     this.topic = topic;
     this.consumer = consumer;
+    this.recordHandler = recordHandler;
 
     workerThread = new Thread(this, "ExampleConsumer Worker-Thread");
     workerThread.start();
@@ -91,6 +94,7 @@ public class ExampleConsumer<K, V> implements Runnable
   {
     consumed++;
     log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);
+    recordHandler.handleRecord(topic, partition, offset, key, value);
   }
 
 

diff --git a/src/main/java/de/juplo/kafka/RecordHandler.java b/src/main/java/de/juplo/kafka/RecordHandler.java
new file mode 100644 (file)
index 0000000..a7b65af
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/RecordHandler.java
@@ -0,0 +1,11 @@
+package de.juplo.kafka;
+
+public interface RecordHandler<K, V>
+{
+  void handleRecord(
+    String topic,
+    Integer partition,
+    Long offset,
+    K key,
+    V value);
+}