juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/SimpleConsumer.java

   1 package de.juplo.kafka;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.clients.consumer.ConsumerRecord;
   5 import org.apache.kafka.clients.consumer.ConsumerRecords;
   6 import org.apache.kafka.clients.consumer.KafkaConsumer;
   7 import org.apache.kafka.common.serialization.StringDeserializer;
   8
   9 import java.time.Duration;
  10 import java.util.Arrays;
  11 import java.util.Properties;
  12
  13
  14 @Slf4j
  15 public class SimpleConsumer
  16 {
  17   private final String id;
  18   private final String topic;
  19   private final KafkaConsumer<String, String> consumer;
  20
  21   private long consumed = 0;
  22   private volatile boolean running = true;
  23   private volatile boolean done = false;
  24
  25   public SimpleConsumer(String broker, String topic, String groupId, String clientId)
  26   {
  27     Properties props = new Properties();
  28     props.put("bootstrap.servers", broker);
  29     props.put("group.id", groupId); // ID für die Offset-Commits
  30     props.put("client.id", clientId); // Nur zur Wiedererkennung
  31     props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
  32     props.put("key.deserializer", StringDeserializer.class.getName());
  33     props.put("value.deserializer", StringDeserializer.class.getName());
  34
  35     consumer = new KafkaConsumer<>(props);
  36
  37     this.topic = topic;
  38     this.id = clientId;
  39   }
  40
  41
  42   public void run()
  43   {
  44     try
  45     {
  46       log.info("{} - Subscribing to topic test", id);
  47       consumer.subscribe(Arrays.asList("test"));
  48
  49       while (running)
  50       {
  51         ConsumerRecords<String, String> records =
  52             consumer.poll(Duration.ofSeconds(1));
  53
  54         log.info("{} - Received {} messages", id, records.count());
  55         for (ConsumerRecord<String, String> record : records)
  56         {
  57           consumed++;
  58           log.info(
  59               "{} - {}: {}/{} - {}={}",
  60               id,
  61               record.offset(),
  62               record.topic(),
  63               record.partition(),
  64               record.key(),
  65               record.value()
  66           );
  67         }
  68       }
  69     }
  70     catch(Exception e)
  71     {
  72       log.error("{} - Unexpected error: {}", id, e.toString());
  73     }
  74     finally
  75     {
  76       log.info("{} - Closing the KafkaConsumer", id);
  77       consumer.close();
  78       log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
  79       done = true;
  80     }
  81   }
  82
  83
  84   public static void main(String[] args) throws Exception
  85   {
  86     String broker = ":9092";
  87     String topic = "test";
  88     String groupId = "my-group";
  89     String clientId = "DEV";
  90
  91     switch (args.length)
  92     {
  93       case 4:
  94         clientId = args[3];
  95       case 3:
  96         groupId = args[2];
  97       case 2:
  98         topic = args[1];
  99       case 1:
 100         broker = args[0];
 101     }
 102
 103
 104     SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId);
 105
 106     Runtime.getRuntime().addShutdownHook(new Thread(() ->
 107     {
 108       instance.running = false;
 109       while (!instance.done)
 110       {
 111         log.info("Waiting for main-thread...");
 112         try
 113         {
 114           Thread.sleep(1000);
 115         }
 116         catch (InterruptedException e) {}
 117       }
 118       log.info("Shutdown completed.");
 119     }));
 120
 121     log.info(
 122         "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}",
 123         broker,
 124         topic,
 125         groupId,
 126         clientId);
 127     instance.run();
 128   }
 129 }