Log-Meldungen für das Senden des Zählerstands ergänzt

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 2dfc2a4..8226f9f 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -231,12 +231,22 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
   public void onPartitionsAssigned(Collection<TopicPartition> partitions)
   {
     phaser.bulkRegister(partitions.size());
+    log.info(
+      "{} - Added {} parties for newly assigned partitions. New total number of parties: {}",
+      id,
+      partitions.size(),
+      phaser.getRegisteredParties());
   }
 
   @Override
   public void onPartitionsRevoked(Collection<TopicPartition> partitions)
   {
     partitions.forEach(partition -> phaser.arriveAndDeregister());
+    log.info(
+      "{} - Removed {} parties for revoked partitions. New total number of parties: {}",
+      id,
+      partitions.size(),
+      phaser.getRegisteredParties());
   }
 
   public void shutdown() throws InterruptedException