`simple-producer` in `simple-consumer` verwandelt -- ALIGN

diff --git a/Dockerfile b/Dockerfile
index 74e66ed..22819af 100644 (file)
--- a/Dockerfile
+++ b/Dockerfile
@@ -3,4 +3,4 @@ VOLUME /tmp
 COPY target/*.jar /opt/app.jar
 COPY target/libs /opt/libs
 ENTRYPOINT [ "java", "-jar", "/opt/app.jar" ]
-CMD [ "kafka:9092", "test", "DCKR" ]
+CMD [ "kafka:9092", "test", "my-group", "DCKR" ]

diff --git a/README.sh b/README.sh
index 3d98ace..85b8f96 100755 (executable)
--- a/README.sh
+++ b/README.sh
@@ -1,6 +1,6 @@
 #!/bin/bash
 
-IMAGE=juplo/simple-producer:1.0-SNAPSHOT
+IMAGE=juplo/simple-consumer:1.0-SNAPSHOT
 
 if [ "$1" = "cleanup" ]
 then
@@ -10,7 +10,7 @@ then
 fi
 
 docker compose -f docker/docker-compose.yml up -d --remove-orphans kafka-1 kafka-2 kafka-3
-docker compose -f docker/docker-compose.yml rm -svf producer
+docker compose -f docker/docker-compose.yml rm -svf consumer
 
 if [[
   $(docker image ls -q $IMAGE) == "" ||
@@ -27,10 +27,13 @@ docker compose -f docker/docker-compose.yml up --remove-orphans setup || exit 1
 
 
 docker compose -f docker/docker-compose.yml up -d producer
+docker compose -f docker/docker-compose.yml up -d consumer
+
 sleep 5
+docker compose -f docker/docker-compose.yml stop consumer
 
-docker compose -f docker/docker-compose.yml exec cli kafkacat -b kafka:9092 -t test -c 20 -f'topic=%t\tpartition=%p\toffset=%o\tkey=%k\tvalue=%s\n'
+docker compose -f docker/docker-compose.yml start consumer
+sleep 5
 
-docker compose -f docker/docker-compose.yml stop producer
-docker compose -f docker/docker-compose.yml exec cli kafkacat -b kafka:9092 -t test -e -f'topic=%t\tpartition=%p\toffset=%o\tkey=%k\tvalue=%s\n'
-docker compose -f docker/docker-compose.yml logs producer
+docker compose -f docker/docker-compose.yml stop producer consumer
+docker compose -f docker/docker-compose.yml logs consumer

diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
index 80f0aec..cba608b 100644 (file)
--- a/docker/docker-compose.yml
+++ b/docker/docker-compose.yml
@@ -193,6 +193,10 @@ services:
     image: juplo/simple-producer:1.0-SNAPSHOT
     command: kafka:9092 test producer
 
+  consumer:
+    image: juplo/simple-consumer:1.0-SNAPSHOT
+    command: kafka:9092 test my-group consumer
+
 volumes:
   zookeeper-data:
   zookeeper-log:

diff --git a/pom.xml b/pom.xml
index ad7f17a..2d81d24 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -12,9 +12,9 @@
   </parent>
 
   <groupId>de.juplo.kafka</groupId>
-  <artifactId>simple-producer</artifactId>
-  <name>Super Simple Producer</name>
-  <description>A Simple Producer, programmed with pure Java, that sends messages via Kafka</description>
+  <artifactId>simple-consumer</artifactId>
+  <name>Simple Consumer-Group</name>
+  <description>Super Simple Consumer-Group, that is implemented as a plain Java-program</description>
   <version>1.0-SNAPSHOT</version>
 
   <properties>
@@ -62,7 +62,7 @@
             <manifest>
               <addClasspath>true</addClasspath>
               <classpathPrefix>libs/</classpathPrefix>
-              <mainClass>de.juplo.kafka.ExampleProducer</mainClass>
+              <mainClass>de.juplo.kafka.ExampleConsumer</mainClass>
             </manifest>
           </archive>
         </configuration>

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index aa014a6..55eb60f 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -1,118 +1,94 @@
 package de.juplo.kafka;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.StringSerializer;
-
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.Consumer;
+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;
 
 
 @Slf4j
-public class ExampleProducer
+public class ExampleConsumer
 {
   private final String id;
   private final String topic;
-  private final Producer<String, String> producer;
+  private final Consumer<String, String> consumer;
 
-  private volatile boolean running = true;
-  private volatile boolean done = false;
-  private long produced = 0;
+  private volatile boolean running = false;
+  private long consumed = 0;
 
-  public ExampleProducer(
+  public ExampleConsumer(
     String broker,
     String topic,
+    String groupId,
     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.serializer", StringSerializer.class.getName());
-    props.put("value.serializer", StringSerializer.class.getName());
-    props.put("metadata.maxage.ms", 5000);
+    props.put("auto.offset.reset", "earliest"); // Von Beginn an lesen
+    props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
+    props.put("key.deserializer", StringDeserializer.class.getName());
+    props.put("value.deserializer", StringDeserializer.class.getName());
 
     this.id = clientId;
     this.topic = topic;
-    producer = new KafkaProducer<>(props);
+    consumer = new KafkaConsumer<>(props);
   }
 
+
   public void run()
   {
-    long i = 0;
-
     try
     {
-      for (; running; i++)
+      log.info("{} - Subscribing to topic {}", id, topic);
+      consumer.subscribe(Arrays.asList(topic));
+      running = true;
+
+      while (true)
       {
-        send(Long.toString(i%10), Long.toString(i));
-        Thread.sleep(500);
+        ConsumerRecords<String, String> records =
+            consumer.poll(Duration.ofSeconds(1));
+
+        log.info("{} - Received {} messages", id, records.count());
+        for (ConsumerRecord<String, String> record : records)
+        {
+          consumed++;
+          log.info(
+              "{} - {}: {}/{} - {}={}",
+              id,
+              record.offset(),
+              record.topic(),
+              record.partition(),
+              record.key(),
+              record.value()
+          );
+        }
       }
     }
-    catch (Exception e)
+    catch(WakeupException e)
     {
-      log.error("{} - Unexpected error!", id, e);
+      log.info("{} - Consumer was signaled to finish its work", id);
     }
-    finally
+    catch(Exception e)
     {
-      log.info("{}: Closing the KafkaProducer", id);
-      producer.close();
-      log.info("{}: Produced {} messages in total, exiting!", id, produced);
-      done = true;
+      log.error("{} - Unexpected error, unsubscribing!", id, e.toString());
+      consumer.unsubscribe();
     }
-  }
-
-  void send(String key, String value)
-  {
-    final long time = System.currentTimeMillis();
-
-    final ProducerRecord<String, String> record = new ProducerRecord<>(
-        topic,  // Topic
-        key,    // Key
-        value   // Value
-    );
-
-    producer.send(record, (metadata, e) ->
+    finally
     {
-      long now = System.currentTimeMillis();
-      if (e == null)
-      {
-        // HANDLE SUCCESS
-        produced++;
-        log.debug(
-            "{} - Sent message {}={}, partition={}:{}, timestamp={}, latency={}ms",
-            id,
-            record.key(),
-            record.value(),
-            metadata.partition(),
-            metadata.offset(),
-            metadata.timestamp(),
-            now - time
-        );
-      }
-      else
-      {
-        // HANDLE ERROR
-        log.error(
-            "{} - ERROR for message {}={}, timestamp={}, latency={}ms: {}",
-            id,
-            record.key(),
-            record.value(),
-            metadata == null ? -1 : metadata.timestamp(),
-            now - time,
-            e.toString()
-        );
-      }
-    });
-
-    long now = System.currentTimeMillis();
-    log.trace(
-        "{} - Queued message {}={}, latency={}ms",
-        id,
-        record.key(),
-        record.value(),
-        now - time
-    );
+      running = false;
+      log.info("{} - Closing the KafkaConsumer", id);
+      consumer.close();
+      log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
+    }
   }
 
 
@@ -120,24 +96,29 @@ public class ExampleProducer
   {
     String broker = ":9092";
     String topic = "test";
+    String groupId = "my-group";
     String clientId = "DEV";
 
     switch (args.length)
     {
+      case 4:
+        clientId = args[3];
       case 3:
-        clientId = args[2];
+        groupId = args[2];
       case 2:
         topic = args[1];
       case 1:
         broker = args[0];
     }
 
-    ExampleProducer instance = new ExampleProducer(broker, topic, clientId);
+
+    ExampleConsumer instance = new ExampleConsumer(broker, topic, groupId, clientId);
 
     Runtime.getRuntime().addShutdownHook(new Thread(() ->
     {
-      instance.running = false;
-      while (!instance.done)
+      instance.consumer.wakeup();
+
+      while (instance.running)
       {
         log.info("Waiting for main-thread...");
         try
@@ -150,9 +131,10 @@ public class ExampleProducer
     }));
 
     log.info(
-        "Running ExampleProducer: broker={}, topic={}, client-id={}",
+        "Running ExampleConsumer: broker={}, topic={}, group-id={}, client-id={}",
         broker,
         topic,
+        groupId,
         clientId);
     instance.run();
   }