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.kafka.annotation.KafkaListener;
7 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
10 import java.util.Optional;
13 @RequiredArgsConstructor
15 public class EndlessConsumer<K, V>
17 private final String id;
18 private final KafkaListenerEndpointRegistry registry;
19 private final ApplicationErrorHandler errorHandler;
20 private final RecordHandler<K, V> recordHandler;
22 private long consumed = 0;
26 id = "${spring.kafka.client-id}",
28 topics = "${sumup.adder.topic}",
30 autoStartup = "false")
31 public void accept(List<ConsumerRecord<K, V>> records)
33 // Do something with the data...
34 log.info("{} - Received {} messages", id, records.size());
35 for (ConsumerRecord<K, V> record : records)
38 "{} - {}: {}/{} - {}={}",
47 recordHandler.accept(record);
56 throw new IllegalStateException("Consumer instance " + id + " is already running!");
58 log.info("{} - Starting - consumed {} messages before", id, consumed);
59 errorHandler.clearState();
60 registry.getListenerContainer(id).start();
66 throw new IllegalStateException("Consumer instance " + id + " is not running!");
68 log.info("{} - Stopping", id);
69 registry.getListenerContainer(id).stop();
70 log.info("{} - Stopped - consumed {} messages so far", id, consumed);
73 public boolean running()
75 return registry.getListenerContainer(id).isRunning();
78 public Optional<Exception> exitStatus()
81 throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
83 return errorHandler.getException();