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.stereotype.Component;
11 import java.util.List;
12 import java.util.function.Consumer;
17 @RequiredArgsConstructor
18 public class EndlessConsumer<K, V>
20 @Value("${consumer.client-id}")
23 Consumer<ConsumerRecord<K, V>> handler;
27 id = "${consumer.client-id}",
29 topics = "${consumer.topic}",
30 containerFactory = "batchFactory",
31 autoStartup = "false")
32 public void receive(List<ConsumerRecord<K, V>> records)
34 // Do something with the data...
35 log.info("{} - Received {} messages", id, records.size());
36 for (ConsumerRecord<K, V> record : records)
39 "{} - {}: {}/{} - {}={}",
48 handler.accept(record);