From c90558e63a9ba9e38a9a0ab3e056fcaa775f4c6f Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 29 Sep 2024 14:30:42 +0200
Subject: [PATCH] Handling der Records wie in abgeleiteten Versionen in eigener
Methode
---
.../java/de/juplo/kafka/ExampleConsumer.java | 27 ++++++++++++-------
1 file changed, 17 insertions(+), 10 deletions(-)
diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index e604c008..7fbd5de9 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -61,16 +61,12 @@ public class ExampleConsumer
log.info("{} - Received {} messages", id, records.count());
for (ConsumerRecord<String, String> record : records)
{
- consumed++;
- log.info(
- "{} - {}: {}/{} - {}={}",
- id,
- record.offset(),
- record.topic(),
- record.partition(),
- record.key(),
- record.value()
- );
+ handleRecord(
+ record.topic(),
+ record.partition(),
+ record.offset(),
+ record.key(),
+ record.value());
}
}
}
@@ -92,6 +88,17 @@ public class ExampleConsumer
}
}
+ private void handleRecord(
+ String topic,
+ Integer partition,
+ Long offset,
+ String key,
+ String value)
+ {
+ consumed++;
+ log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value);
+ }
+
public static void main(String[] args) throws Exception
{
--
2.20.1