X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=delivery%2Fsrc%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Foutbox%2Fdelivery%2FOutboxProducer.java;h=331374df1d71727344107a4245f2472c4a6b7400;hb=a92d318043bf698dc2a949db2b76893c8abf03a1;hp=09c8789a3ac25697f70cabcb146473e90d5e5ecd;hpb=70ec0ac46ecafcd10cb4224c67439f9039067bba;p=demos%2Fkafka%2Foutbox

diff --git a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
index 09c8789..331374d 100644
--- a/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
+++ b/delivery/src/main/java/de/juplo/kafka/outbox/delivery/OutboxProducer.java
@@ -1,40 +1,52 @@
 package de.juplo.kafka.outbox.delivery;
 
 import com.google.common.primitives.Longs;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
+import java.time.Clock;
 import java.time.Duration;
-import java.util.List;
-import java.util.Properties;
+import java.time.LocalTime;
+import java.util.*;
 
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.scheduling.annotation.Scheduled;
-import org.springframework.stereotype.Component;
 
 import javax.annotation.PreDestroy;
 
 import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
+import static org.apache.kafka.clients.consumer.ConsumerConfig.*;
 import static org.apache.kafka.clients.producer.ProducerConfig.*;
 
 
-@Component
 public class OutboxProducer
 {
   final static Logger LOG = LoggerFactory.getLogger(OutboxProducer.class);
 
+  public final static String HEADER = "#";
 
   private final OutboxRepository repository;
   private final KafkaProducer<String, String> producer;
   private final String topic;
+  private final Watermarks watermarks;
+  private final Clock clock;
+  private final Duration cleanupInterval;
 
   private long sequenceNumber = 0l;
+  private LocalTime nextCleanup;
 
   public OutboxProducer(
       ApplicationProperties properties,
-      OutboxRepository repository)
+      OutboxRepository repository,
+      Clock clock)
   {
     this.repository = repository;
 
@@ -46,6 +58,67 @@ public class OutboxProducer
 
     this.producer = new KafkaProducer<>(props);
     this.topic = properties.topic;
+
+    props = new Properties();
+    props.put(BOOTSTRAP_SERVERS_CONFIG, properties.bootstrapServers);
+    props.put(GROUP_ID_CONFIG, "outbox");
+    props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+    props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+
+    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+    consumer.subscribe(Arrays.asList(this.topic));
+    List<PartitionInfo> partitions = consumer.listTopics().get(this.topic);
+    Set<TopicPartition> assignment = new HashSet<>();
+    for (PartitionInfo info : partitions)
+    {
+      LOG.debug("Found {}/{} (ISR: {})", info.topic(), info.partition(), info.inSyncReplicas());
+      assignment.add(new TopicPartition(info.topic(), info.partition()));
+    }
+
+    LOG.info("Using topic {} with {} partitions", topic, partitions);
+
+    this.watermarks = new Watermarks(partitions.size());
+
+    long[] currentOffsets = new long[partitions.size()];
+    for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : consumer.committed(assignment).entrySet())
+    {
+      LOG.info("Found current offset {} for partition {}", entry.getValue(), entry.getKey());
+      currentOffsets[entry.getKey().partition()] = entry.getValue().offset() - 1l;
+    }
+    LOG.info("Current offsets: {}", currentOffsets);
+
+    long[] endOffsets = new long[partitions.size()];
+    for (Map.Entry<TopicPartition, Long> entry : consumer.endOffsets(assignment).entrySet())
+    {
+      LOG.info("Found next offset {} for partition {}", entry.getValue(), entry.getKey());
+      endOffsets[entry.getKey().partition()] = entry.getValue() - 1l;
+    }
+    LOG.info("End-offsets: {}", endOffsets);
+
+    while(!Arrays.equals(currentOffsets, endOffsets))
+    {
+      ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
+      LOG.debug("Fetched {} records", records.count());
+      records.forEach(record ->
+      {
+        long recordSequenceNumber = Longs.fromByteArray(record.headers().lastHeader(HEADER).value());
+        LOG.debug("Found watermark partition[{}]={}", record.partition(), recordSequenceNumber);
+        watermarks.set(record.partition(), recordSequenceNumber);
+        currentOffsets[record.partition()] = record.offset();
+      });
+      LOG.debug("Current offsets: {}", currentOffsets);
+    }
+
+    LOG.info("Found watermarks: {}", watermarks);
+
+    sequenceNumber = watermarks.getLowest();
+    LOG.info("Restored sequence-number: {}", sequenceNumber);
+
+    consumer.close();
+
+    this.clock = clock;
+    this.cleanupInterval = properties.cleanupInterval;
+    this.nextCleanup = LocalTime.now(clock);
   }
 
   @Scheduled(fixedDelayString = "${de.juplo.kafka.outbox.interval}")
@@ -58,6 +131,15 @@ public class OutboxProducer
       LOG.debug("Polled {} new items", items.size());
       for (OutboxItem item : items)
         send(item);
+      if (nextCleanup.isBefore(LocalTime.now(clock)))
+      {
+        int deleted = repository.delete(watermarks.getLowest());
+        nextCleanup = LocalTime.now(clock).plus(cleanupInterval);
+        LOG.info(
+            "Cleaned up {} entries from outbox, next clean-up: {}",
+            deleted,
+            nextCleanup);
+      }
     }
     while (items.size() > 0);
   }
@@ -68,22 +150,21 @@ public class OutboxProducer
         new ProducerRecord<>(topic, item.getKey(), item.getValue());
 
     sequenceNumber = item.getSequenceNumber();
-    record.headers().add("SEQ#", Longs.toByteArray(sequenceNumber));
+    record.headers().add(HEADER, Longs.toByteArray(sequenceNumber));
 
     producer.send(record, (metadata, e) ->
     {
       if (metadata != null)
       {
-        int deleted = repository.delete(item.getSequenceNumber());
+        watermarks.set(metadata.partition(), item.getSequenceNumber());
         LOG.info(
-            "{}/{}:{} - {}:{}={} - deleted: {}",
+            "{}/{}:{} - {}:{}={}",
             metadata.topic(),
             metadata.partition(),
             metadata.offset(),
             item.getSequenceNumber(),
             record.key(),
-            record.value(),
-            deleted);
+            record.value());
       }
       else
       {