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.function.Consumer;
  13
  14
  15 @Component
  16 @Slf4j
  17 @RequiredArgsConstructor
  18 public class EndlessConsumer<K, V>
  19 {
  20   @Autowired
  21   private KafkaListenerEndpointRegistry registry;
  22   @Value("${consumer.client-id}")
  23   String id;
  24   @Autowired
  25   Consumer<ConsumerRecord<K, V>> handler;
  26
  27   private long consumed = 0;
  28
  29   @KafkaListener(
  30       id = "${consumer.client-id}",
  31       idIsGroup = false,
  32       topics = "${consumer.topic}",
  33       autoStartup = "false")
  34   public void receive(ConsumerRecord<K, V> record)
  35   {
  36     log.info(
  37         "{} - {}: {}/{} - {}={}",
  38         id,
  39         record.offset(),
  40         record.topic(),
  41         record.partition(),
  42         record.key(),
  43         record.value()
  44     );
  45
  46     handler.accept(record);
  47
  48     consumed++;
  49   }
  50
  51
  52   public synchronized void start()
  53   {
  54     if (registry.getListenerContainer(id).isChildRunning())
  55       throw new IllegalStateException("Consumer instance " + id + " is already running!");
  56
  57     log.info("{} - Starting - consumed {} messages before", id, consumed);
  58     registry.getListenerContainer(id).start();
  59   }
  60
  61   public synchronized void stop()
  62   {
  63     if (!registry.getListenerContainer(id).isChildRunning())
  64       throw new IllegalStateException("Consumer instance " + id + " is not running!");
  65
  66     log.info("{} - Stopping", id);
  67     registry.getListenerContainer(id).stop();
  68     log.info("{} - Stopped - consumed {} messages so far", id, consumed);
  69   }
  70 }