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.errors.WakeupException;
   8 import org.apache.kafka.common.serialization.StringDeserializer;
   9
  10 import java.time.Duration;
  11 import java.util.Arrays;
  12 import java.util.Properties;
  13 import java.util.concurrent.locks.Condition;
  14 import java.util.concurrent.locks.Lock;
  15 import java.util.concurrent.locks.ReentrantLock;
  16
  17
  18 @Slf4j
  19 public class SimpleConsumer
  20 {
  21   private long consumed = 0;
  22   private KafkaConsumer<String, String> consumer;
  23   private Lock lock = new ReentrantLock();
  24   private Condition stopped = lock.newCondition();
  25
  26
  27   public SimpleConsumer()
  28   {
  29     // tag::create[]
  30     Properties props = new Properties();
  31     props.put("bootstrap.servers", ":9092");
  32     props.put("group.id", "my-consumer"); // << Used for Offset-Management
  33     // end::create[]
  34     props.put("auto.offset.reset", "earliest");
  35     // tag::create[]
  36     props.put("key.deserializer", StringDeserializer.class.getName());
  37     props.put("value.deserializer", StringDeserializer.class.getName());
  38
  39     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  40     // end::create[]
  41     this.consumer = consumer;
  42   }
  43
  44
  45   public void run()
  46   {
  47     String id = "C";
  48
  49     try
  50     {
  51       log.info("{} - Subscribing to topic test", id);
  52       consumer.subscribe(Arrays.asList("test"));
  53
  54       // tag::loop[]
  55       while (true)
  56       {
  57         ConsumerRecords<String, String> records =
  58             consumer.poll(Duration.ofSeconds(1));
  59
  60         // Do something with the data...
  61         // end::loop[]
  62         log.info("{} - Received {} messages", id, records.count());
  63         for (ConsumerRecord<String, String> record : records)
  64         {
  65           consumed++;
  66           log.info(
  67               "{} - {}: {}/{} - {}={}",
  68               id,
  69               record.offset(),
  70               record.topic(),
  71               record.partition(),
  72               record.key(),
  73               record.value()
  74           );
  75         }
  76         // tag::loop[]
  77       }
  78       // end::loop[]
  79     }
  80     catch(WakeupException e)
  81     {
  82       log.info("{} - RIIING!", id);
  83     }
  84     catch(Exception e)
  85     {
  86       log.error("{} - Unexpected error: {}", id, e.toString());
  87     }
  88     finally
  89     {
  90       this.lock.lock();
  91       try
  92       {
  93         log.info("{} - Closing the KafkaConsumer", id);
  94         consumer.close();
  95         log.info("{} - DONE!");
  96         stopped.signal();
  97       }
  98       finally
  99       {
 100         this.lock.unlock();
 101         log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
 102       }
 103     }
 104   }
 105
 106
 107   public static void main(String[] args) throws Exception
 108   {
 109     SimpleConsumer instance = new SimpleConsumer();
 110
 111     Runtime.getRuntime().addShutdownHook(new Thread(() ->
 112     {
 113       instance.lock.lock();
 114       try
 115       {
 116         instance.consumer.wakeup();
 117         instance.stopped.await();
 118       }
 119       catch (InterruptedException e)
 120       {
 121         log.warn("Interrrupted while waiting for the consumer to stop!", e);
 122       }
 123       finally
 124       {
 125         instance.lock.unlock();
 126       }
 127     }));
 128
 129     instance.run();
 130   }
 131 }