From 2bd5b7ca99188f72952156c389b545ab471ab936 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 2 Aug 2022 19:39:55 +0200 Subject: [PATCH] Vorlage --- .../java/de/juplo/kafka/SimpleConsumer.java | 24 +++++-------------- 1 file changed, 6 insertions(+), 18 deletions(-) diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java index cab2fb8..d5290bb 100644 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java @@ -45,28 +45,16 @@ public class SimpleConsumer try { log.info("{} - Subscribing to topic test", id); - consumer.subscribe(Arrays.asList("test")); + // TODO: subscribe! running = true; while (true) { - ConsumerRecords records = - consumer.poll(Duration.ofSeconds(1)); - - 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() - ); - } + // TODO: + // * Poll for new records + // * Loop through received records + // * Count received records in the variable `consumed` + // * "Work" with the data: Log partition/offset/key/value } } catch(WakeupException e) -- 2.20.1