juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/SimpleConsumer.java

   1 package de.juplo.kafka;
   2
   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;
  10
  11
  12 @Slf4j
  13 @Component
  14 public class SimpleConsumer
  15 {
  16   @Value("${spring.kafka.client-id}")
  17   private String id;
  18   private long consumed = 0;
  19
  20   @KafkaListener(topics = "${simple.consumer.topic}")
  21   private void handleRecord(
  22     @Header(KafkaHeaders.RECEIVED_TOPIC)
  23     String topic,
  24     @Header(KafkaHeaders.RECEIVED_PARTITION_ID)
  25     Integer partition,
  26     @Header(KafkaHeaders.OFFSET)
  27     Long offset,
  28     @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)
  29     String key,
  30     @Payload
  31     String value)
  32   {
  33     consumed++;
  34     log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value);
  35   }
  36 }