From: Kai Moritz <kai@juplo.de>
Date: Sun, 13 Apr 2025 09:12:38 +0000 (+0200)
Subject: Handling der Nachricht in das Interface `RecordHandler` verlegt
X-Git-Tag: consumer/spring-consumer--record-handler--2026-03-21--smartlifecycle-only
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=9102c30c6dacfdad7808ff18349e56a378056ba7;p=demos%2Fkafka%2Ftraining

Handling der Nachricht in das Interface `RecordHandler` verlegt
---

diff --git a/README.sh b/README.sh
index a7cbb894..203c22bd 100755
--- 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 3f628abf..983dff98 100644
--- 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 852d49a2..849889bf 100644
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -180,7 +180,7 @@ services:
     mem_limit:  100m
 
   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
@@ -189,7 +189,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
@@ -198,7 +198,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 cace641e..31dfb2f2 100644
--- 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 6d271817..840a793f 100644
--- 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.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.kafka.autoconfigure.KafkaProperties;
@@ -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)
@@ -24,7 +27,14 @@ public class ApplicationConfiguration
       new ExampleConsumer<>(
         kafkaProperties.getClientId(),
         properties.getConsumerProperties().getTopic(),
-        kafkaConsumer);
+        kafkaConsumer,
+        recordHandler);
+  }
+
+  @Bean
+  public RecordHandler<String, String> recordHandler()
+  {
+    return (topic, partition, offset, key, value) -> log.info("No-Ops Handler called for {}={}", key, value);
   }
 
   @Bean(destroyMethod = "")
diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index f3368641..9c63d537 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -17,6 +17,7 @@ public class ExampleConsumer<K, V> implements Runnable, SmartLifecycle
   private final String id;
   private final String topic;
   private final Consumer<K, V> consumer;
+  private final RecordHandler<K, V> recordHandler;
 
   private Thread workerThread;
   private volatile boolean running = false;
@@ -26,11 +27,13 @@ public class ExampleConsumer<K, V> implements Runnable, SmartLifecycle
   public ExampleConsumer(
     String clientId,
     String topic,
-    Consumer<K, V> consumer)
+    Consumer<K, V> consumer,
+    RecordHandler<K, V> recordHandler)
   {
     this.id = clientId;
     this.topic = topic;
     this.consumer = consumer;
+    this.recordHandler = recordHandler;
   }
 
   @Override
@@ -97,6 +100,7 @@ public class ExampleConsumer<K, V> implements Runnable, SmartLifecycle
   {
     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
index 00000000..a7b65af2
--- /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);
+}