juplo.de Git - demos/spring/data-jdbc/blob - src/main/java/de/trion/kafka/outbox/OutboxConsumer.java

   1 package de.trion.kafka.outbox;
   2
   3 import com.fasterxml.jackson.databind.ObjectMapper;
   4 import org.apache.kafka.clients.consumer.ConsumerRecord;
   5 import org.apache.kafka.clients.consumer.ConsumerRecords;
   6 import org.apache.kafka.clients.consumer.KafkaConsumer;
   7 import org.apache.kafka.common.errors.WakeupException;
   8 import org.apache.kafka.common.serialization.LongDeserializer;
   9 import org.apache.kafka.common.serialization.StringDeserializer;
  10 import org.slf4j.Logger;
  11 import org.slf4j.LoggerFactory;
  12 import org.springframework.boot.ApplicationArguments;
  13 import org.springframework.boot.ApplicationRunner;
  14 import org.springframework.stereotype.Component;
  15
  16 import javax.annotation.PreDestroy;
  17 import java.time.Duration;
  18 import java.util.Arrays;
  19 import java.util.Properties;
  20
  21
  22 @Component
  23 public class OutboxConsumer implements ApplicationRunner, Runnable {
  24
  25     private final static Logger LOG = LoggerFactory.getLogger(OutboxConsumer.class);
  26
  27     private final OutboxService service;
  28     private final OutboxProducer sender;
  29     private final ObjectMapper mapper;
  30     private final String topic;
  31     private final KafkaConsumer<Long, String> consumer;
  32     private final Thread thread;
  33
  34     private long internalState = 1;
  35
  36
  37     public OutboxConsumer(
  38             OutboxService service,
  39             OutboxProducer sender,
  40             ObjectMapper mapper,
  41             String bootstrapServers,
  42             String consumerGroup,
  43             String topic) {
  44
  45         this.service = service;
  46         this.sender = sender;
  47         this.mapper = mapper;
  48         this.topic = topic;
  49
  50         Properties props = new Properties();
  51         props.put("bootstrap.servers", bootstrapServers);
  52         props.put("group.id", consumerGroup);
  53         props.put("auto.commit.interval.ms", 15000);
  54         props.put("metadata.max.age.ms", 1000);
  55         props.put("key.deserializer", LongDeserializer.class.getName());
  56         props.put("value.deserializer", StringDeserializer.class.getName());
  57         consumer = new KafkaConsumer<>(props);
  58
  59         thread = new Thread(this);
  60     }
  61
  62
  63     @Override
  64     public void run() {
  65         try
  66         {
  67             LOG.info("Subscribing to topic " + topic);
  68             consumer.subscribe(Arrays.asList(topic));
  69
  70             while (true)
  71             {
  72                 ConsumerRecords<Long, String> records = consumer.poll(Duration.ofSeconds(1));
  73                 for (ConsumerRecord<Long, String> record : records) {
  74                     LOG.debug("Ignoring command {}", record.value());
  75                 }
  76             }
  77         }
  78         catch (WakeupException e) {}
  79         catch (Exception e) {
  80             LOG.error("Unexpected exception!", e);
  81         }
  82         finally
  83         {
  84             LOG.info("Closing the KafkaConsumer...");
  85             try {
  86                 consumer.close(Duration.ofSeconds(5));
  87                 LOG.debug("Successfully closed the KafkaConsumer");
  88             }
  89             catch (Exception e) {
  90                 LOG.warn("Exception while closing the KafkaConsumer!", e);
  91             }
  92         }
  93     }
  94
  95
  96     @Override
  97     public void run(ApplicationArguments args) {
  98         thread.start();
  99         try {
 100             thread.join();
 101             LOG.info("Successfully joined the consumer-thread");
 102         }
 103         catch (InterruptedException e) {
 104             LOG.info("Main-thread was interrupted while joining the consumer-thread");
 105         }
 106     }
 107
 108     @PreDestroy
 109     public void stop()
 110     {
 111         LOG.info("Stopping the KafkaConsumer...");
 112         consumer.wakeup();
 113     }
 114 }