From c90558e63a9ba9e38a9a0ab3e056fcaa775f4c6f Mon Sep 17 00:00:00 2001 From: Kai Moritz 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 e604c00..7fbd5de 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 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