From 6ff05112077432d7538d269e48f8bd49b5a394bc Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 22 Nov 2022 19:29:32 +0100 Subject: [PATCH] Vorlage --- pom.xml | 4 --- .../java/de/juplo/kafka/SimpleConsumer.java | 25 +++++-------------- 2 files changed, 6 insertions(+), 23 deletions(-) diff --git a/pom.xml b/pom.xml index 985797b..a07cfaa 100644 --- a/pom.xml +++ b/pom.xml @@ -34,10 +34,6 @@ - - pl.project13.maven - git-commit-id-plugin - org.apache.maven.plugins maven-dependency-plugin diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java index 9741e67..880ed78 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 {}", id, topic); - consumer.subscribe(Arrays.asList(topic)); + // 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) @@ -76,7 +64,6 @@ public class SimpleConsumer catch(Exception e) { log.error("{} - Unexpected error: {}, unsubscribing!", id, e.toString()); - consumer.unsubscribe(); } finally { -- 2.20.1