From ba21d6becd37dbe82fa1e71b1b44ac2589bc661a Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 1 Aug 2022 22:55:48 +0200 Subject: [PATCH] Implementierung des Consumers aus 'first-contact' vereinfacht MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Code für ordentliches Herunterfahren so weit wie möglich vereinfacht. * Dabei den Code möglichst eng an die Producer-Implementierung angelehnt. --- pom.xml | 4 +- .../java/de/juplo/kafka/SimpleConsumer.java | 104 ++++++++-------- .../java/de/juplo/kafka/SimpleProducer.java | 117 ------------------ 3 files changed, 55 insertions(+), 170 deletions(-) delete mode 100644 src/main/java/de/juplo/kafka/SimpleProducer.java diff --git a/pom.xml b/pom.xml index 70f37e8..2b39e97 100644 --- a/pom.xml +++ b/pom.xml @@ -12,8 +12,8 @@ de.juplo.kafka - first-contact - First Contact: a Simple Producer and a simple Consumer-Group + simple-consumer + Super Simple Consumer-Group 1.0-SNAPSHOT diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java index eaa9566..4e2ca8c 100644 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ b/src/main/java/de/juplo/kafka/SimpleConsumer.java @@ -4,55 +4,49 @@ import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import java.time.Duration; import java.util.Arrays; import java.util.Properties; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; @Slf4j public class SimpleConsumer { - private long consumed = 0; - private KafkaConsumer consumer; - private Lock lock = new ReentrantLock(); - private Condition stopped = lock.newCondition(); + private final String id; + private final String topic; + private final KafkaConsumer consumer; + private long consumed = 0; + private volatile boolean running = true; + private volatile boolean done = false; - public SimpleConsumer() + public SimpleConsumer(String broker, String topic, String groupId, String clientId) { - // tag::create[] Properties props = new Properties(); - props.put("bootstrap.servers", ":9092"); - props.put("group.id", "my-consumer"); // << Used for Offset-Commits - // end::create[] - props.put("auto.offset.reset", "earliest"); - // tag::create[] + 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()); - KafkaConsumer consumer = new KafkaConsumer<>(props); - // end::create[] - this.consumer = consumer; + consumer = new KafkaConsumer<>(props); + + this.topic = topic; + this.id = clientId; } public void run() { - String id = "C"; - try { log.info("{} - Subscribing to topic test", id); consumer.subscribe(Arrays.asList("test")); // tag::loop[] - while (true) + while (running) { ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); @@ -77,55 +71,63 @@ public class SimpleConsumer } // end::loop[] } - catch(WakeupException e) - { - log.info("{} - RIIING!", id); - } catch(Exception e) { log.error("{} - Unexpected error: {}", id, e.toString()); } finally { - this.lock.lock(); - try - { - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("{} - DONE!"); - stopped.signal(); - } - finally - { - this.lock.unlock(); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); - } + log.info("{} - Closing the KafkaConsumer", id); + consumer.close(); + log.info("{}: Consumed {} messages in total, exiting!", id, consumed); + done = true; } } public static void main(String[] args) throws Exception { - SimpleConsumer instance = new SimpleConsumer(); + String broker = ":9092"; + String topic = "test"; + String groupId = "my-group"; + String clientId = "DEV"; + + switch (args.length) + { + case 4: + clientId = args[3]; + case 3: + groupId = args[2]; + case 2: + topic = args[1]; + case 1: + broker = args[0]; + } + + + SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId); Runtime.getRuntime().addShutdownHook(new Thread(() -> { - instance.lock.lock(); - try - { - instance.consumer.wakeup(); - instance.stopped.await(); - } - catch (InterruptedException e) - { - log.warn("Interrrupted while waiting for the consumer to stop!", e); - } - finally + instance.running = false; + while (!instance.done) { - instance.lock.unlock(); + log.info("Waiting for main-thread..."); + try + { + Thread.sleep(1000); + } + catch (InterruptedException e) {} } + log.info("Shutdown completed."); })); + log.info( + "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}", + broker, + topic, + groupId, + clientId); instance.run(); } } diff --git a/src/main/java/de/juplo/kafka/SimpleProducer.java b/src/main/java/de/juplo/kafka/SimpleProducer.java deleted file mode 100644 index 43a7227..0000000 --- a/src/main/java/de/juplo/kafka/SimpleProducer.java +++ /dev/null @@ -1,117 +0,0 @@ -package de.juplo.kafka; - -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.producer.KafkaProducer; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.StringSerializer; - -import java.util.Properties; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; - - -@Slf4j -public class SimpleProducer -{ - private final String id; - private final String topic; - private final KafkaProducer producer; - - private long produced = 0; - - public SimpleProducer(String clientId, String topic) - { - // tag::create[] - Properties props = new Properties(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); - - KafkaProducer producer = new KafkaProducer<>(props); - // end::create[] - - this.id = clientId; - this.topic = topic; - this.producer = producer; - } - - public void run() - { - long i = 0; - - try - { - for (; i < 100 ; i++) - { - send(Long.toString(i%10), Long.toString(i)); - } - - log.info("{} - Done", id); - } - finally - { - log.info("{}: Closing the KafkaProducer", id); - producer.close(); - log.info("{}: Produced {} messages in total, exiting!", id, produced); - } - } - - void send(String key, String value) - { - final long time = System.currentTimeMillis(); - - final ProducerRecord record = new ProducerRecord<>( - topic, // Topic - key, // Key - value // Value - ); - - producer.send(record, (metadata, e) -> - { - long now = System.currentTimeMillis(); - if (e == null) - { - // HANDLE SUCCESS - produced++; - log.debug( - "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms", - id, - record.key(), - record.value(), - metadata.partition(), - metadata.offset(), - metadata.timestamp(), - now - time - ); - } - else - { - // HANDLE ERROR - log.error( - "{} - ERROR key={} timestamp={} latency={}ms: {}", - id, - record.key(), - metadata == null ? -1 : metadata.timestamp(), - now - time, - e.toString() - ); - } - }); - - long now = System.currentTimeMillis(); - log.trace( - "{} - Queued #{} key={} latency={}ms", - id, - value, - record.key(), - now - time - ); - } - - - public static void main(String[] args) throws Exception - { - SimpleProducer producer = new SimpleProducer("P", "test"); - producer.run(); - } -} -- 2.20.1