X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;h=9e75112516bb9129720bcbedbcd4c045889c0b3b;hb=12c1ce703ef76b75d995f94d1689c894dde1406a;hp=6d3850f677b33e8ad71695527ac9386e3e42c1fd;hpb=3fcb695db13281056c541a10a08abff4d162495c;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index 6d3850f..9e75112 100644
--- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@ -2,6 +2,7 @@ package de.juplo.kafka;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
 
@@ -9,12 +10,14 @@ import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.*;
+import java.util.concurrent.CountDownLatch;
 
 
 @RequiredArgsConstructor
 @Slf4j
 public class ApplicationRebalanceListener implements ConsumerRebalanceListener
 {
+  private final Consumer consumer;
   private final ApplicationRecordHandler recordHandler;
   private final AdderResults adderResults;
   private final StateRepository stateRepository;
@@ -51,6 +54,24 @@ public class ApplicationRebalanceListener implements ConsumerRebalanceListener
   @Override
   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
   {
+    log.info("{} - Commiting offsets for all previously assigned partitions", id);
+    CountDownLatch commitDone = new CountDownLatch(1);
+    consumer.commitAsync((offsets, e) ->
+    {
+      commitDone.countDown();
+      if (e == null)
+      {
+        log.error("{} - Could not commit offsets to Kafka!", id, e);
+      }
+      else
+      {
+        offsets.entrySet().stream().forEach(entry ->
+        {
+          log.info("{} - Commited offset for {}: {}", id, entry.getKey(), entry.getValue());
+        });
+      }
+    });
+
     partitions.forEach(tp ->
     {
       Integer partition = tp.partition();
@@ -69,5 +90,19 @@ public class ApplicationRebalanceListener implements ConsumerRebalanceListener
       Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
       stateRepository.save(new StateDocument(partition, state, results));
     });
+
+    try
+    {
+      log.debug("{} - Waiting for async commit to complete", id);
+      commitDone.await();
+    }
+    catch (InterruptedException e)
+    {
+      log.warn(
+        "{} - Interrupted while waiting for async commit in onPartitionsRevoked({})",
+        id,
+        partitions,
+        e);
+    }
   }
 }