1 package de.trion.kafka.outbox;
3 import com.fasterxml.jackson.databind.ObjectMapper;
4 import de.lvm.tx.Event;
5 import de.lvm.tx.Command;
6 import de.lvm.tx.Command.Action;
7 import org.apache.kafka.clients.consumer.ConsumerRecord;
8 import org.apache.kafka.clients.consumer.ConsumerRecords;
9 import org.apache.kafka.clients.consumer.KafkaConsumer;
10 import org.apache.kafka.common.errors.WakeupException;
11 import org.apache.kafka.common.serialization.LongDeserializer;
12 import org.apache.kafka.common.serialization.StringDeserializer;
13 import org.slf4j.Logger;
14 import org.slf4j.LoggerFactory;
15 import org.springframework.boot.ApplicationArguments;
16 import org.springframework.boot.ApplicationRunner;
17 import org.springframework.stereotype.Component;
19 import javax.annotation.PreDestroy;
21 import java.io.IOException;
22 import java.nio.charset.StandardCharsets;
23 import java.time.Duration;
24 import java.util.Arrays;
25 import java.util.Properties;
27 import static de.lvm.tx.Event.Type.*;
30 public class OutboxConsumer implements ApplicationRunner, Runnable {
32 private final static Logger LOG = LoggerFactory.getLogger(OutboxConsumer.class);
34 private final OutboxService service;
35 private final OutboxProducer sender;
36 private final ObjectMapper mapper;
37 private final String topic;
38 private final KafkaConsumer<Long, String> consumer;
39 private final Thread thread;
41 private long internalState = 1;
44 public OutboxConsumer(
45 OutboxService service,
46 OutboxProducer sender,
48 String bootstrapServers,
52 this.service = service;
57 Properties props = new Properties();
58 props.put("bootstrap.servers", bootstrapServers);
59 props.put("group.id", consumerGroup);
60 props.put("auto.commit.interval.ms", 15000);
61 props.put("metadata.max.age.ms", 1000);
62 props.put("key.deserializer", LongDeserializer.class.getName());
63 props.put("value.deserializer", StringDeserializer.class.getName());
64 consumer = new KafkaConsumer<>(props);
66 thread = new Thread(this);
74 LOG.info("Subscribing to topic " + topic);
75 consumer.subscribe(Arrays.asList(topic));
79 ConsumerRecords<Long, String> records = consumer.poll(Duration.ofSeconds(1));
80 for (ConsumerRecord<Long, String> record : records) {
81 byte code = record.headers().lastHeader("messageType").value()[0];
82 Action action = Action.from(code);
86 LOG.debug("Ignoring unknown action {} for {}", code, record.value());
92 dlzSaveReceived(toCommand(record.value()));
95 LOG.debug("Ignoring message {}", record.value());
97 byte[] bytes = record.headers().lastHeader("messageType").value();
98 String type = new String(bytes, StandardCharsets.UTF_8);
100 if (type.endsWith("DlzAction")) {
101 dlzSaveReceived(toCommand(record.value()));
105 LOG.debug("Ignoring command {}", record.value());
109 catch (WakeupException e) {}
110 catch (Exception e) {
111 LOG.error("Unexpected exception!", e);
115 LOG.info("Closing the KafkaConsumer...");
117 consumer.close(Duration.ofSeconds(5));
118 LOG.debug("Successfully closed the KafkaConsumer");
120 catch (Exception e) {
121 LOG.warn("Exception while closing the KafkaConsumer!", e);
126 public Command toCommand(String message) throws IOException {
127 Command command = mapper.readValue(message, Command.class);
128 LOG.info("{}: {}", command.getAction(), command.getVorgangId());
132 public void dlzSaveReceived(Command command) throws InterruptedException {
136 service.bearbeiteVorgang(
137 command.getVorgangId(),
140 reply(command, result);
142 catch (Exception e) {
143 LOG.error("Exception during processing!", e);
147 public void reply(Command command, String message) {
148 String vorgangId = command.getVorgangId();
149 String vbId = command.getVbId();
150 Event event = new Event(DLZ_SAVED, vorgangId, vbId);
151 event.getZustand().put(Event.DLZ, message);
157 public void run(ApplicationArguments args) {
161 LOG.info("Successfully joined the consumer-thread");
163 catch (InterruptedException e) {
164 LOG.info("Main-thread was interrupted while joining the consumer-thread");
171 LOG.info("Stopping the KafkaConsumer...");