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);
29 private final OutboxRepository repository;
30 private final KafkaProducer<String, String> producer;
31 private final String topic;
32 private final Watermarks watermarks;
33 private final Clock clock;
34 private final Duration cleanupInterval;
36 private long sequenceNumber = 0l;
37 private LocalTime nextCleanup;
39 public OutboxProducer(
40 ApplicationProperties properties,
41 OutboxRepository repository,
44 this.repository = repository;
46 Properties props = new Properties();
47 props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
48 props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
49 props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
50 props.put(ENABLE_IDEMPOTENCE_CONFIG, true);
52 this.producer = new KafkaProducer<>(props);
53 this.topic = properties.topic;
55 this.watermarks = new Watermarks();
57 this.cleanupInterval = properties.cleanupInterval;
58 this.nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
61 @Scheduled(fixedDelayString = "${de.juplo.kafka.outbox.interval}")
64 List<OutboxItem> items;
67 items = repository.fetch(sequenceNumber);
68 LOG.debug("Polled {} new items", items.size());
69 for (OutboxItem item : items)
71 if (nextCleanup.isBefore(LocalTime.now(clock)))
73 int deleted = repository.delete(watermarks.getLowest());
74 nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
76 "Cleaned up {} entries from outbox, next clean-up: {}",
81 while (items.size() > 0);
84 void send(OutboxItem item)
86 final ProducerRecord<String, String> record =
87 new ProducerRecord<>(topic, item.getKey(), item.getValue());
89 sequenceNumber = item.getSequenceNumber();
90 record.headers().add("SEQ#", Longs.toByteArray(sequenceNumber));
92 producer.send(record, (metadata, e) ->
96 watermarks.set(metadata.partition(), item.getSequenceNumber());
98 "{}/{}:{} - {}:{}={}",
100 metadata.partition(),
102 item.getSequenceNumber(),
110 "{}/{} - {}:{}={} -> ",
113 item.getSequenceNumber(),
125 producer.close(Duration.ofSeconds(5));