1 package de.juplo.kafka.outbox.delivery;
3 import com.google.common.primitives.Longs;
4 import org.apache.kafka.common.serialization.StringSerializer;
6 import java.time.Duration;
8 import java.util.Properties;
10 import org.apache.kafka.clients.producer.KafkaProducer;
11 import org.apache.kafka.clients.producer.ProducerRecord;
12 import org.slf4j.Logger;
13 import org.slf4j.LoggerFactory;
14 import org.springframework.scheduling.annotation.Scheduled;
15 import org.springframework.stereotype.Component;
17 import javax.annotation.PreDestroy;
19 import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
20 import static org.apache.kafka.clients.producer.ProducerConfig.*;
24 public class OutboxProducer
26 final static Logger LOG = LoggerFactory.getLogger(OutboxProducer.class);
29 private final OutboxRepository repository;
30 private final KafkaProducer<String, String> producer;
31 private final String topic;
33 private long sequenceNumber = 0l;
35 public OutboxProducer(
36 ApplicationProperties properties,
37 OutboxRepository repository)
39 this.repository = repository;
41 Properties props = new Properties();
42 props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
43 props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
44 props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
45 props.put(ENABLE_IDEMPOTENCE_CONFIG, true);
47 this.producer = new KafkaProducer<>(props);
48 this.topic = properties.topic;
51 @Scheduled(fixedDelayString = "${de.juplo.kafka.outbox.interval}")
54 List<OutboxItem> items;
57 items = repository.fetch(sequenceNumber);
58 LOG.debug("Polled {} new items", items.size());
59 for (OutboxItem item : items)
62 while (items.size() > 0);
65 void send(OutboxItem item)
67 final ProducerRecord<String, String> record =
68 new ProducerRecord<>(topic, item.getKey(), item.getValue());
70 sequenceNumber = item.getSequenceNumber();
71 record.headers().add("SEQ#", Longs.toByteArray(sequenceNumber));
73 producer.send(record, (metadata, e) ->
77 int deleted = repository.delete(item.getSequenceNumber());
79 "{}/{}:{} - {}:{}={} - deleted: {}",
83 item.getSequenceNumber(),
92 "{}/{} - {}:{}={} -> ",
95 item.getSequenceNumber(),
107 producer.close(Duration.ofSeconds(5));