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(
  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)
  26     String topic,
  27     @Header(KafkaHeaders.RECEIVED_PARTITION_ID)
  28     Integer partition,
  29     @Header(KafkaHeaders.OFFSET)
  30     Long offset,
  31     @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)
  32     String key,
  33     @Payload
  34     String value)
  35   {
  36     consumed++;
  37     log.info("{} - {}: {}/{} - {}={}", id, offset, topic, partition, key, value);
  38   }
  39 }