1 package de.juplo.kafka;
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;
12 import javax.annotation.PreDestroy;
13 import java.util.List;
14 import java.util.function.Consumer;
19 @RequiredArgsConstructor
20 public class EndlessConsumer<K, V>
23 private KafkaListenerEndpointRegistry registry;
24 @Value("${consumer.client-id}")
27 Consumer<ConsumerRecord<K, V>> handler;
29 private long consumed = 0;
32 id = "${consumer.client-id}",
34 topics = "${consumer.topic}",
35 containerFactory = "batchFactory",
36 autoStartup = "false")
37 public void receive(List<ConsumerRecord<K, V>> records)
39 // Do something with the data...
40 log.info("{} - Received {} messages", id, records.size());
41 for (ConsumerRecord<K, V> record : records)
44 "{} - {}: {}/{} - {}={}",
53 handler.accept(record);
60 public synchronized void start()
62 if (registry.getListenerContainer(id).isChildRunning())
63 throw new IllegalStateException("Consumer instance " + id + " is already running!");
65 log.info("{} - Starting - consumed {} messages before", id, consumed);
66 registry.getListenerContainer(id).start();
69 public synchronized void stop()
71 if (!registry.getListenerContainer(id).isChildRunning())
72 throw new IllegalStateException("Consumer instance " + id + " is not running!");
74 log.info("{} - Stopping", id);
75 registry.getListenerContainer(id).stop();
76 log.info("{} - Stopped - consumed {} messages so far", id, consumed);
82 log.info("{} - Destroy!", id);
87 catch (IllegalStateException e)
89 log.info("{} - Was already stopped", id);
93 log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
97 log.info("{}: Consumed {} messages in total, exiting!", id, consumed);