From: Kai Moritz <kai@juplo.de>
Date: Mon, 28 Oct 2024 10:13:05 +0000 (+0100)
Subject: Fehler im Logging der aktiven Phase korrigiert und Meldungen verbessert
X-Git-Tag: consumer/spring-consumer--log-compaction--2024-11-13--si~17
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=1cbfef1ef5f134b5f1eaa0de1bfda3bbd4b59441;p=demos%2Fkafka%2Ftraining

Fehler im Logging der aktiven Phase korrigiert und Meldungen verbessert
---

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 3acaebe..385df67 100644
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -81,7 +81,9 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
         ConsumerRecords<String, String> records =
             consumer.poll(Duration.ofSeconds(1));
 
-        log.info("{} - Received {} messages", id, records.count());
+        int phase = phaser.getPhase();
+
+        log.info("{} - Received {} messages in phase {}", id, records.count(), phase);
         records
           .partitions()
           .forEach(partition ->
@@ -103,8 +105,8 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
             done[partition.partition()] = true;
           });
 
-        int phase = phaser.arriveAndAwaitAdvance();
-        log.info("{} - Phase {} is done!", id, phase);
+        int arrivedPhase = phaser.arriveAndAwaitAdvance();
+        log.info("{} - Phase {} is done! Next phase: {}", id, phase, arrivedPhase);
       }
     }
     catch(WakeupException e)