From 6c2c6e4d75e4b4ff1f7220f3d77f895db93e8e85 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 2 Feb 2025 22:14:31 +0100
Subject: [PATCH] `@KafkaListener` mit paraleller Verarbeitung

---
 src/main/java/de/juplo/kafka/ExampleConsumer.java | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 5894a103..58a249e0 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -18,7 +18,7 @@ public class ExampleConsumer
   private String id;
   private long consumed = 0;
 
-  @KafkaListener(topics = "${juplo.consumer.topic}")
+  @KafkaListener(topics = "${juplo.consumer.topic}", concurrency = "2")
   private void receive(
     @Header(KafkaHeaders.RECEIVED_TOPIC)
     String topic,
@@ -41,6 +41,9 @@ public class ExampleConsumer
     String key,
     String value)
   {
+    // BEACHTE:
+    // Der Zugriff auf die Variable "consumed" ist hier so nicht mehr korrekt!
+    // Grund: Der Zugriff erfolgt ohne Locking konkurrierend aus zwei Threads.
     consumed++;
     log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value);
   }
-- 
2.20.1