1 package de.juplo.kafka.outbox.delivery;
3 import com.google.common.primitives.Longs;
4 import org.apache.kafka.clients.consumer.ConsumerRecords;
5 import org.apache.kafka.clients.consumer.KafkaConsumer;
6 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
7 import org.apache.kafka.common.PartitionInfo;
8 import org.apache.kafka.common.TopicPartition;
9 import org.apache.kafka.common.serialization.StringDeserializer;
10 import org.apache.kafka.common.serialization.StringSerializer;
12 import java.time.Clock;
13 import java.time.Duration;
14 import java.time.LocalTime;
17 import org.apache.kafka.clients.producer.KafkaProducer;
18 import org.apache.kafka.clients.producer.ProducerRecord;
19 import org.slf4j.Logger;
20 import org.slf4j.LoggerFactory;
21 import org.springframework.scheduling.annotation.Scheduled;
23 import javax.annotation.PreDestroy;
25 import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
26 import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
27 import static org.apache.kafka.clients.producer.ProducerConfig.*;
30 public class OutboxProducer
32 final static Logger LOG = LoggerFactory.getLogger(OutboxProducer.class);
34 public final static String HEADER = "#";
36 private final OutboxRepository repository;
37 private final KafkaProducer<String, String> producer;
38 private final String topic;
39 private final Watermarks watermarks;
40 private final Clock clock;
41 private final Duration cleanupInterval;
43 private long sequenceNumber = 0l;
44 private LocalTime nextCleanup;
46 public OutboxProducer(
47 ApplicationProperties properties,
48 OutboxRepository repository,
51 this.repository = repository;
53 Properties props = new Properties();
54 props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
55 props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
56 props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
57 props.put(ENABLE_IDEMPOTENCE_CONFIG, true);
59 this.producer = new KafkaProducer<>(props);
60 this.topic = properties.topic;
62 props = new Properties();
63 props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
64 props.put(GROUP_ID_CONFIG, "outbox");
65 props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
66 props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
68 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
69 consumer.subscribe(Arrays.asList(this.topic));
70 List<PartitionInfo> partitions = consumer.listTopics().get(this.topic);
71 Set<TopicPartition> assignment = new HashSet<>();
72 for (PartitionInfo info : partitions)
74 LOG.debug("Found {}/{} (ISR: {})", info.topic(), info.partition(), info.inSyncReplicas());
75 assignment.add(new TopicPartition(info.topic(), info.partition()));
78 LOG.info("Using topic {} with {} partitions", topic, partitions);
80 this.watermarks = new Watermarks(partitions.size());
82 long[] currentOffsets = new long[partitions.size()];
83 for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : consumer.committed(assignment).entrySet())
85 LOG.info("Found current offset {} for partition {}", entry.getValue(), entry.getKey());
86 currentOffsets[entry.getKey().partition()] = entry.getValue().offset() - 1l;
88 LOG.info("Current offsets: {}", currentOffsets);
90 long[] endOffsets = new long[partitions.size()];
91 for (Map.Entry<TopicPartition, Long> entry : consumer.endOffsets(assignment).entrySet())
93 LOG.info("Found next offset {} for partition {}", entry.getValue(), entry.getKey());
94 endOffsets[entry.getKey().partition()] = entry.getValue() - 1l;
96 LOG.info("End-offsets: {}", endOffsets);
98 while(!Arrays.equals(currentOffsets, endOffsets))
100 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
101 LOG.debug("Fetched {} records", records.count());
102 records.forEach(record ->
104 long recordSequenceNumber = Longs.fromByteArray(record.headers().lastHeader(HEADER).value());
105 LOG.debug("Found watermark partition[{}]={}", record.partition(), recordSequenceNumber);
106 watermarks.set(record.partition(), recordSequenceNumber);
107 currentOffsets[record.partition()] = record.offset();
109 LOG.debug("Current offsets: {}", currentOffsets);
112 LOG.info("Found watermarks: {}", watermarks);
114 sequenceNumber = watermarks.getLowest();
115 LOG.info("Restored sequence-number: {}", sequenceNumber);
120 this.cleanupInterval = properties.cleanupInterval;
121 this.nextCleanup = LocalTime.now(clock);
124 @Scheduled(fixedDelayString = "${de.juplo.kafka.outbox.interval}")
127 List<OutboxItem> items;
130 items = repository.fetch(sequenceNumber);
131 LOG.debug("Polled {} new items", items.size());
132 for (OutboxItem item : items)
134 if (nextCleanup.isBefore(LocalTime.now(clock)))
136 int deleted = repository.delete(watermarks.getLowest());
137 nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
139 "Cleaned up {} entries from outbox, next clean-up: {}",
144 while (items.size() > 0);
147 void send(OutboxItem item)
149 final ProducerRecord<String, String> record =
150 new ProducerRecord<>(topic, item.getKey(), item.getValue());
152 sequenceNumber = item.getSequenceNumber();
153 record.headers().add(HEADER, Longs.toByteArray(sequenceNumber));
155 producer.send(record, (metadata, e) ->
157 if (metadata != null)
159 watermarks.set(metadata.partition(), item.getSequenceNumber());
161 "{}/{}:{} - {}:{}={}",
163 metadata.partition(),
165 item.getSequenceNumber(),
173 "{}/{} - {}:{}={} -> ",
176 item.getSequenceNumber(),
188 producer.close(Duration.ofSeconds(5));