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

   1 package de.juplo.kafka;
   2
   3 import lombok.RequiredArgsConstructor;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.clients.consumer.*;
   6 import org.apache.kafka.common.TopicPartition;
   7 import org.apache.kafka.common.errors.RecordDeserializationException;
   8 import org.apache.kafka.common.errors.WakeupException;
   9 import org.springframework.context.ConfigurableApplicationContext;
  10
  11 import java.time.Duration;
  12 import java.util.*;
  13 import java.util.concurrent.ExecutorService;
  14
  15
  16 @Slf4j
  17 @RequiredArgsConstructor
  18 public class EndlessConsumer<K, V> implements Runnable
  19 {
  20   private final ExecutorService executor;
  21   private final ConfigurableApplicationContext applicationContext;
  22   private final String id;
  23   private final String topic;
  24   private final Consumer<K, V> consumer;
  25   private final ConsumerRebalanceListener rebalanceListener;
  26   private final RecordHandler<K, V> recordHandler;
  27
  28   private boolean running = false;
  29   private Exception exception = null;
  30   private long consumed = 0;
  31
  32
  33
  34   @Override
  35   public void run()
  36   {
  37     try
  38     {
  39       log.info("{} - Subscribing to topic {}", id, topic);
  40       consumer.subscribe(Arrays.asList(topic), rebalanceListener);
  41
  42       while (true)
  43       {
  44         ConsumerRecords<K, V> records =
  45             consumer.poll(Duration.ofSeconds(1));
  46
  47         // Do something with the data...
  48         log.info("{} - Received {} messages", id, records.count());
  49         for (ConsumerRecord<K, V> record : records)
  50         {
  51           log.info(
  52               "{} - {}: {}/{} - {}={}",
  53               id,
  54               record.offset(),
  55               record.topic(),
  56               record.partition(),
  57               record.key(),
  58               record.value()
  59           );
  60
  61           recordHandler.accept(record);
  62
  63           consumed++;
  64         }
  65       }
  66     }
  67     catch(WakeupException e)
  68     {
  69       log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id);
  70     }
  71     catch(RecordDeserializationException e)
  72     {
  73       TopicPartition tp = e.topicPartition();
  74       long offset = e.offset();
  75       log.error(
  76           "{} - Could not deserialize  message on topic {} with offset={}: {}",
  77           id,
  78           tp,
  79           offset,
  80           e.getCause().toString());
  81       this.exception = e;
  82     }
  83     catch(Exception e)
  84     {
  85       log.error("{} - Unexpected error: {}", id, e.toString(), e);
  86       this.exception = e;
  87       log.info("{} - Unsubscribing...", id);
  88       consumer.unsubscribe();
  89     }
  90     finally
  91     {
  92       running = false;
  93       log.info("{} - Closing the consumer...", id);
  94       consumer.close();
  95       log.info("{} - Shutting down the app...", id);
  96       applicationContext.close();
  97       log.info("{} - Consumer-Thread exiting", id);
  98     }
  99   }
 100
 101   public void start()
 102   {
 103     if (running)
 104       throw new IllegalStateException("Consumer instance " + id + " is already running!");
 105
 106     log.info("{} - Starting - consumed {} messages before", id, consumed);
 107     running = true;
 108     executor.submit(this);
 109   }
 110
 111   public void stop()
 112   {
 113     consumer.wakeup();
 114   }
 115   public boolean running()
 116   {
 117     return running;
 118   }
 119
 120   public Optional<Exception> exitStatus()
 121   {
 122     if (running)
 123       throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
 124
 125     return Optional.ofNullable(exception);
 126   }
 127 }