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 java.util.Optional;
13 import java.util.function.Consumer;
18 @RequiredArgsConstructor
19 public class EndlessConsumer<K, V>
22 private KafkaListenerEndpointRegistry registry;
23 @Value("${consumer.client-id}")
26 Consumer<ConsumerRecord<K, V>> handler;
28 ApplicationErrorHandler errorHandler;
30 private long consumed = 0;
33 id = "${consumer.client-id}",
35 topics = "${consumer.topic}",
36 autoStartup = "false")
37 public void receive(ConsumerRecord<K, V> record)
40 "{} - {}: {}/{} - {}={}",
49 handler.accept(record);
55 public synchronized void start()
57 if (registry.getListenerContainer(id).isChildRunning())
58 throw new IllegalStateException("Consumer instance " + id + " is already running!");
60 log.info("{} - Starting - consumed {} messages before", id, consumed);
61 errorHandler.clearException();
62 registry.getListenerContainer(id).start();
65 public synchronized void stop()
67 if (!registry.getListenerContainer(id).isChildRunning())
68 throw new IllegalStateException("Consumer instance " + id + " is not running!");
70 log.info("{} - Stopping", id);
71 registry.getListenerContainer(id).stop();
72 log.info("{} - Stopped - consumed {} messages so far", id, consumed);
75 public synchronized Optional<Exception> exitStatus()
77 if (registry.getListenerContainer(id).isChildRunning())
78 throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
80 return errorHandler.getException();