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.ConsumerRecord;
   6 import org.springframework.beans.factory.annotation.Autowired;
   7 import org.springframework.beans.factory.annotation.Value;
   8 import org.springframework.kafka.annotation.KafkaListener;
   9 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
  10 import org.springframework.stereotype.Component;
  11
  12 import java.util.Optional;
  13 import java.util.function.Consumer;
  14
  15
  16 @Component
  17 @Slf4j
  18 @RequiredArgsConstructor
  19 public class EndlessConsumer<K, V>
  20 {
  21   @Autowired
  22   private KafkaListenerEndpointRegistry registry;
  23   @Value("${consumer.client-id}")
  24   String id;
  25   @Autowired
  26   Consumer<ConsumerRecord<K, V>> handler;
  27   @Autowired
  28   ApplicationErrorHandler errorHandler;
  29
  30   private long consumed = 0;
  31
  32   @KafkaListener(
  33       id = "${consumer.client-id}",
  34       idIsGroup = false,
  35       topics = "${consumer.topic}",
  36       autoStartup = "false")
  37   public void receive(ConsumerRecord<K, V> record)
  38   {
  39     log.info(
  40         "{} - {}: {}/{} - {}={}",
  41         id,
  42         record.offset(),
  43         record.topic(),
  44         record.partition(),
  45         record.key(),
  46         record.value()
  47     );
  48
  49     handler.accept(record);
  50
  51     consumed++;
  52   }
  53
  54
  55   public synchronized void start()
  56   {
  57     if (registry.getListenerContainer(id).isChildRunning())
  58       throw new IllegalStateException("Consumer instance " + id + " is already running!");
  59
  60     log.info("{} - Starting - consumed {} messages before", id, consumed);
  61     errorHandler.clearException();
  62     registry.getListenerContainer(id).start();
  63   }
  64
  65   public synchronized void stop()
  66   {
  67     if (!registry.getListenerContainer(id).isChildRunning())
  68       throw new IllegalStateException("Consumer instance " + id + " is not running!");
  69
  70     log.info("{} - Stopping", id);
  71     registry.getListenerContainer(id).stop();
  72     log.info("{} - Stopped - consumed {} messages so far", id, consumed);
  73   }
  74
  75   public synchronized Optional<Exception> exitStatus()
  76   {
  77     if (registry.getListenerContainer(id).isChildRunning())
  78       throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
  79
  80     return errorHandler.getException();
  81   }
  82 }