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.Clock;
7 import java.time.Duration;
8 import java.time.LocalTime;
10 import java.util.Properties;
12 import org.apache.kafka.clients.producer.KafkaProducer;
13 import org.apache.kafka.clients.producer.ProducerRecord;
14 import org.slf4j.Logger;
15 import org.slf4j.LoggerFactory;
16 import org.springframework.scheduling.annotation.Scheduled;
18 import javax.annotation.PreDestroy;
20 import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
21 import static org.apache.kafka.clients.producer.ProducerConfig.*;
24 public class OutboxProducer
26 final static Logger LOG = LoggerFactory.getLogger(OutboxProducer.class);
28 public final static String HEADER = "#";
30 private final OutboxRepository repository;
31 private final KafkaProducer<String, String> producer;
32 private final String topic;
33 private final Watermarks watermarks;
34 private final Clock clock;
35 private final Duration cleanupInterval;
37 private long sequenceNumber = 0l;
38 private LocalTime nextCleanup;
40 public OutboxProducer(
41 ApplicationProperties properties,
42 OutboxRepository repository,
45 this.repository = repository;
47 Properties props = new Properties();
48 props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
49 props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
50 props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
51 props.put(ENABLE_IDEMPOTENCE_CONFIG, true);
53 this.producer = new KafkaProducer<>(props);
54 this.topic = properties.topic;
56 this.watermarks = new Watermarks();
58 this.cleanupInterval = properties.cleanupInterval;
59 this.nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
62 @Scheduled(fixedDelayString = "${de.juplo.kafka.outbox.interval}")
65 List<OutboxItem> items;
68 items = repository.fetch(sequenceNumber);
69 LOG.debug("Polled {} new items", items.size());
70 for (OutboxItem item : items)
72 if (nextCleanup.isBefore(LocalTime.now(clock)))
74 int deleted = repository.delete(watermarks.getLowest());
75 nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
77 "Cleaned up {} entries from outbox, next clean-up: {}",
82 while (items.size() > 0);
85 void send(OutboxItem item)
87 final ProducerRecord<String, String> record =
88 new ProducerRecord<>(topic, item.getKey(), item.getValue());
90 sequenceNumber = item.getSequenceNumber();
91 record.headers().add(HEADER, Longs.toByteArray(sequenceNumber));
93 producer.send(record, (metadata, e) ->
97 watermarks.set(metadata.partition(), item.getSequenceNumber());
99 "{}/{}:{} - {}:{}={}",
101 metadata.partition(),
103 item.getSequenceNumber(),
111 "{}/{} - {}:{}={} -> ",
114 item.getSequenceNumber(),
126 producer.close(Duration.ofSeconds(5));