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.WakeupException;
   8
   9 import javax.annotation.PreDestroy;
  10 import java.time.Duration;
  11 import java.util.*;
  12 import java.util.concurrent.ExecutionException;
  13 import java.util.concurrent.ExecutorService;
  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 @RequiredArgsConstructor
  21 public class EndlessConsumer implements Runnable
  22 {
  23   private final ExecutorService executor;
  24   private final String id;
  25   private final String topic;
  26   private final Consumer<String, String> consumer;
  27
  28   private final Lock lock = new ReentrantLock();
  29   private final Condition condition = lock.newCondition();
  30   private boolean running = false;
  31   private Exception exception;
  32   private long consumed = 0;
  33
  34   private final Map<Integer, Map<String, Long>> seen = new HashMap<>();
  35   private final Map<Integer, Long> offsets = new HashMap<>();
  36
  37
  38   @Override
  39   public void run()
  40   {
  41     try
  42     {
  43       log.info("{} - Subscribing to topic {}", id, topic);
  44       consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
  45       {
  46         @Override
  47         public void onPartitionsRevoked(Collection<TopicPartition> partitions)
  48         {
  49           partitions.forEach(tp ->
  50           {
  51             Integer partition = tp.partition();
  52             Long newOffset = consumer.position(tp);
  53             Long oldOffset = offsets.remove(partition);
  54             log.info(
  55                 "{} - removing partition: {}, consumed {} records (offset {} -> {})",
  56                 id,
  57                 partition,
  58                 newOffset - oldOffset,
  59                 oldOffset,
  60                 newOffset);
  61             Map<String, Long> removed = seen.remove(partition);
  62             for (String key : removed.keySet())
  63             {
  64               log.info(
  65                   "{} - Seen {} messages for partition={}|key={}",
  66                   id,
  67                   removed.get(key),
  68                   partition,
  69                   key);
  70             }
  71           });
  72         }
  73
  74         @Override
  75         public void onPartitionsAssigned(Collection<TopicPartition> partitions)
  76         {
  77           partitions.forEach(tp ->
  78           {
  79             Integer partition = tp.partition();
  80             Long offset = consumer.position(tp);
  81             log.info("{} - adding partition: {}, offset={}", id, partition, offset);
  82             offsets.put(partition, offset);
  83             seen.put(partition, new HashMap<>());
  84           });
  85         }
  86       });
  87
  88       while (true)
  89       {
  90         ConsumerRecords<String, String> records =
  91             consumer.poll(Duration.ofSeconds(1));
  92
  93         // Do something with the data...
  94         log.info("{} - Received {} messages", id, records.count());
  95         for (ConsumerRecord<String, String> record : records)
  96         {
  97           consumed++;
  98           log.info(
  99               "{} - {}: {}/{} - {}={}",
 100               id,
 101               record.offset(),
 102               record.topic(),
 103               record.partition(),
 104               record.key(),
 105               record.value()
 106           );
 107
 108           Integer partition = record.partition();
 109           String key = record.key() == null ? "NULL" : record.key();
 110           Map<String, Long> byKey = seen.get(partition);
 111
 112           if (!byKey.containsKey(key))
 113             byKey.put(key, 0l);
 114
 115           long seenByKey = byKey.get(key);
 116           seenByKey++;
 117           byKey.put(key, seenByKey);
 118         }
 119       }
 120     }
 121     catch(WakeupException e)
 122     {
 123       log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id);
 124       consumer.commitSync();
 125       shutdown();
 126     }
 127     catch(Exception e)
 128     {
 129       log.error("{} - Unexpected error: {}", id, e.toString(), e);
 130       shutdown(e);
 131     }
 132     finally
 133     {
 134       log.info("{} - Consumer-Thread exiting", id);
 135     }
 136   }
 137
 138   private void shutdown()
 139   {
 140     shutdown(null);
 141   }
 142
 143   private void shutdown(Exception e)
 144   {
 145     lock.lock();
 146     try
 147     {
 148       try
 149       {
 150         log.info("{} - Unsubscribing from topic {}", id, topic);
 151         consumer.unsubscribe();
 152       }
 153       catch (Exception ue)
 154       {
 155         log.error(
 156             "{} - Error while unsubscribing from topic {}: {}",
 157             id,
 158             topic,
 159             ue.toString());
 160       }
 161       finally
 162       {
 163         running = false;
 164         exception = e;
 165         condition.signal();
 166       }
 167     }
 168     finally
 169     {
 170       lock.unlock();
 171     }
 172   }
 173
 174   public Map<Integer, Map<String, Long>> getSeen()
 175   {
 176     return seen;
 177   }
 178
 179   public void start()
 180   {
 181     lock.lock();
 182     try
 183     {
 184       if (running)
 185         throw new IllegalStateException("Consumer instance " + id + " is already running!");
 186
 187       log.info("{} - Starting - consumed {} messages before", id, consumed);
 188       running = true;
 189       exception = null;
 190       executor.submit(this);
 191     }
 192     finally
 193     {
 194       lock.unlock();
 195     }
 196   }
 197
 198   public synchronized void stop() throws ExecutionException, InterruptedException
 199   {
 200     lock.lock();
 201     try
 202     {
 203       if (!running)
 204         throw new IllegalStateException("Consumer instance " + id + " is not running!");
 205
 206       log.info("{} - Stopping", id);
 207       consumer.wakeup();
 208       condition.await();
 209       log.info("{} - Stopped - consumed {} messages so far", id, consumed);
 210     }
 211     finally
 212     {
 213       lock.unlock();
 214     }
 215   }
 216
 217   @PreDestroy
 218   public void destroy() throws ExecutionException, InterruptedException
 219   {
 220     log.info("{} - Destroy!", id);
 221     try
 222     {
 223       stop();
 224     }
 225     catch (IllegalStateException e)
 226     {
 227       log.info("{} - Was already stopped", id);
 228     }
 229     catch (Exception e)
 230     {
 231       log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
 232     }
 233     finally
 234     {
 235       log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
 236     }
 237   }
 238
 239   public boolean running()
 240   {
 241     lock.lock();
 242     try
 243     {
 244       return running;
 245     }
 246     finally
 247     {
 248       lock.unlock();
 249     }
 250   }
 251
 252   public Optional<Exception> exitStatus()
 253   {
 254     lock.lock();
 255     try
 256     {
 257       if (running)
 258         throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
 259
 260       return Optional.ofNullable(exception);
 261     }
 262     finally
 263     {
 264       lock.unlock();
 265     }
 266   }
 267 }