WIP
[demos/kafka/outbox] / 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 }