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.KafkaHandler;
7 import org.springframework.kafka.annotation.KafkaListener;
8 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
9 import org.springframework.kafka.support.KafkaHeaders;
10 import org.springframework.messaging.handler.annotation.Header;
11 import org.springframework.messaging.handler.annotation.Payload;
13 import java.util.List;
14 import java.util.Optional;
17 @RequiredArgsConstructor
20 id = "${spring.kafka.client-id}",
22 topics = "${sumup.adder.topic}",
23 autoStartup = "false")
24 public class EndlessConsumer
26 private final String id;
27 private final KafkaListenerEndpointRegistry registry;
28 private final RecordHandler recordHandler;
30 private long consumed = 0;
34 public void addNumber(
35 @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
36 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
37 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
38 @Header(KafkaHeaders.OFFSET) Long offset,
39 @Payload MessageAddNumber message)
42 "{} - {}: {}/{} - {}={}",
51 recordHandler.addNumber(topic, partition, offset, key, message);
57 public void calculateSum(
58 @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
59 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
60 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
61 @Header(KafkaHeaders.OFFSET) Long offset,
62 @Payload MessageCalculateSum message)
65 "{} - {}: {}/{} - {}={}",
74 recordHandler.calculateSum(topic, partition, offset, key, message);
82 throw new IllegalStateException("Consumer instance " + id + " is already running!");
84 log.info("{} - Starting - consumed {} messages before", id, consumed);
85 registry.getListenerContainer(id).start();
91 throw new IllegalStateException("Consumer instance " + id + " is not running!");
93 log.info("{} - Stopping", id);
94 registry.getListenerContainer(id).stop();
95 log.info("{} - Stopped - consumed {} messages so far", id, consumed);
98 public boolean running()
100 return registry.getListenerContainer(id).isRunning();