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