From: Kai Moritz <kai@juplo.de>
Date: Tue, 23 Aug 2022 15:55:14 +0000 (+0200)
Subject: `commitAsync()` in `onPartitionsRevoked()`
X-Git-Tag: sumup-adder---lvm-2-tage~5^2~5
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=12c1ce703ef76b75d995f94d1689c894dde1406a;p=demos%2Fkafka%2Ftraining

`commitAsync()` in `onPartitionsRevoked()`

* Der Rebalance-Listener führt jetzt einen (zusätzlichen) Commit der
  Offsets durch.
* Ohne diesen Commit, kann es bei sehr hohem Nachrichten-Aufkommen dazu
  kommen, dass nicht die letztendliche Offset-Position gespeichert wird.
* Da jetzt implizit ein Commit in `onPartitionsRevoked()` durchgeführt
  wird, ist kein expliziter Commit mehr nötig, wenn der Consumer durch
  eine Exception unterbrochen wird, bei der sichergestellt ist, dass die
  zuletzt durch `poll()` gelieferten Nachrichten vollständig verarbeitet
  wurden.
** Bei einer `WakeupException` ist dies klar, da diese in `poll()`
   geworfen wurde, also _nachdem_ die Implementierung durch den Aufruf
   von `poll()` signalisiert hat, dass sie alle zuvor gelieferten
   Nachrichten vollständig verarbeitet hat.
** Bei einer `RecordDeserializationException` ist dies klar, da ein
   Fehler während der Deserialisierung der vom Broker empfangenen
   Nachrichten dazu führt, dass die Kafka-Client-Library die zuvor
   fehlerfrei deserialisierten Nachrichten ausliefert und dies auch
   entsprechend in den intern mitgeführten Offset-Positionen reflektiert.
* Der Commit wird hier asynchron durchgeführt.
* TODO: Das führt dazu, dass die Implementierung in dem Rebalance
  einfriert, da der Callback, auf den sie wartet, dort nie aufgerufen
  wird, da die Commit-Callbacks nur "synchron" abgearbeitet werden, wenn
  die `poll()`-Methode aufgerufen wird!
---

diff --git a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
index 4d056c4..cdd587d 100644
--- a/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
+++ b/src/main/java/de/juplo/kafka/ApplicationConfiguration.java
@@ -35,12 +35,14 @@ public class ApplicationConfiguration
 
   @Bean
   public ApplicationRebalanceListener rebalanceListener(
+      KafkaConsumer<String, String> kafkaConsumer,
       ApplicationRecordHandler recordHandler,
       AdderResults adderResults,
       StateRepository stateRepository,
       ApplicationProperties properties)
   {
     return new ApplicationRebalanceListener(
+        kafkaConsumer,
         recordHandler,
         adderResults,
         stateRepository,
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);
+    }
   }
 }
diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 00678c4..f0e74d3 100644
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -71,8 +71,7 @@ public class EndlessConsumer<K, V> implements Runnable
     }
     catch(WakeupException e)
     {
-      log.info("{} - RIIING! Request to stop consumption - commiting current offsets!", id);
-      consumer.commitSync();
+      log.info("{} - RIIING! Request to stop consumption.", id);
       shutdown();
     }
     catch(RecordDeserializationException e)
@@ -86,7 +85,6 @@ public class EndlessConsumer<K, V> implements Runnable
           offset,
           e.getCause().toString());
 
-      consumer.commitSync();
       shutdown(e);
     }
     catch(Exception e)