From a66978002ec0a060d5eb33a6348ec234fe129908 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 24 Jul 2022 16:12:04 +0200 Subject: [PATCH] Fehler in Logging-Ausgabe korrigiert MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Der über den Merge hinzugefügt Test hat einen Fehler aufgedeckt. * In onPartitionsRevoked() wurde bei der Berechnung der verarbeiteten Nachrichten für die Log-Ausgabe ein Nullzeiger dereferenziert. * Ursache dafür war, dass die Map `offsets` in der Version, die die Offsets speichert gar nicht mehr gepflegt wurde. --- src/main/java/de/juplo/kafka/EndlessConsumer.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java index 366a3c2..3132bd2 100644 --- a/src/main/java/de/juplo/kafka/EndlessConsumer.java +++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java @@ -35,7 +35,7 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnabl private long consumed = 0; private final Map> seen = new HashMap<>(); - private final Map offsets = new HashMap<>(); + private final Map lastOffsets = new HashMap<>(); @Override @@ -45,7 +45,7 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnabl { Integer partition = tp.partition(); Long newOffset = consumer.position(tp); - Long oldOffset = offsets.remove(partition); + Long oldOffset = lastOffsets.remove(partition); log.info( "{} - removing partition: {}, consumed {} records (offset {} -> {})", id, @@ -80,6 +80,7 @@ public class EndlessConsumer implements ConsumerRebalanceListener, Runnabl .findById(Integer.toString(partition)) .orElse(new StatisticsDocument(partition)); consumer.seek(tp, document.offset); + lastOffsets.put(partition, document.offset); seen.put(partition, document.statistics); }); } -- 2.20.1