From 6f5a0736fc8d5d10e11b1d8b10cce06c9d8b12d4 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 22 Nov 2022 19:29:32 +0100 Subject: [PATCH] Vorlage --- .../java/de/juplo/kafka/ExampleConsumer.java | 35 +++---------------- 1 file changed, 4 insertions(+), 31 deletions(-) diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java index f9da8250..9213b1c7 100644 --- a/src/main/java/de/juplo/kafka/ExampleConsumer.java +++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java @@ -30,11 +30,7 @@ public class ExampleConsumer String clientId) { Properties props = new Properties(); - props.put("bootstrap.servers", broker); - props.put("group.id", groupId); // ID für die Offset-Commits - props.put("client.id", clientId); // Nur zur Wiedererkennung - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", StringDeserializer.class.getName()); + // Konfiguration für den Consumer zusammenstellen this.id = clientId; this.topic = topic; @@ -47,24 +43,13 @@ public class ExampleConsumer 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) - { - handleRecord( - record.topic(), - record.partition(), - record.offset(), - record.key(), - record.value()); - } + // TODO: + // Über consumer.poll() Nachrichten abrufen und loggen } } catch(WakeupException e) @@ -74,7 +59,6 @@ public class ExampleConsumer catch(Exception e) { log.error("{} - Unexpected error, unsubscribing!", id, e); - consumer.unsubscribe(); } finally { @@ -85,17 +69,6 @@ public class ExampleConsumer } } - private void handleRecord( - String topic, - Integer partition, - Long offset, - String key, - String value) - { - consumed++; - log.info("{} - partition={}-{}, offset={}: {}={}", id, topic, partition, offset, key, value); - } - public static void main(String[] args) throws Exception { -- 2.20.1