1 package de.juplo.kafka;
3 import lombok.extern.slf4j.Slf4j;
4 import org.springframework.beans.factory.annotation.Value;
5 import org.springframework.kafka.annotation.KafkaListener;
6 import org.springframework.kafka.support.KafkaHeaders;
7 import org.springframework.messaging.handler.annotation.Header;
8 import org.springframework.messaging.handler.annotation.Payload;
9 import org.springframework.stereotype.Component;
14 public class SimpleConsumer
16 @Value("${spring.kafka.client-id}")
18 private long consumed = 0;
21 id = "${spring.kafka.client-id}",
22 groupId = "${spring.kafka.consumer.group-id}",
23 topics = "${simple.consumer.topic}")
24 private void handleRecord(
25 @Header(KafkaHeaders.RECEIVED_TOPIC)
27 @Header(KafkaHeaders.RECEIVED_PARTITION_ID)
29 @Header(KafkaHeaders.OFFSET)
31 @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)
37 log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value);