From: Kai Moritz Date: Sun, 14 Aug 2022 16:25:54 +0000 (+0200) Subject: fix: In `onPartitionsAssigned()` wurde der Kafka-Offset ausgegeben X-Git-Tag: sumup-adder---lvm-2-tage~20 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=a2445f04436bb1087c9fbcd44d319496e91a90c8;hp=461c4014c22b0c0600f8d08578f5676f12d3b036;p=demos%2Fkafka%2Ftraining fix: In `onPartitionsAssigned()` wurde der Kafka-Offset ausgegeben --- diff --git a/src/main/java/de/juplo/kafka/AdderRebalanceListener.java b/src/main/java/de/juplo/kafka/AdderRebalanceListener.java index 7526929..ef595ba 100644 --- a/src/main/java/de/juplo/kafka/AdderRebalanceListener.java +++ b/src/main/java/de/juplo/kafka/AdderRebalanceListener.java @@ -32,12 +32,11 @@ public class AdderRebalanceListener implements PollIntervalAwareConsumerRebalanc partitions.forEach(tp -> { Integer partition = tp.partition(); - Long offset = consumer.position(tp); - log.info("{} - adding partition: {}, offset={}", id, partition, offset); StateDocument document = repository .findById(Integer.toString(partition)) .orElse(new StateDocument(partition)); + log.info("{} - adding partition: {}, offset={}", id, partition, document.offset); if (document.offset >= 0) { // Only seek, if a stored offset was found @@ -54,15 +53,15 @@ public class AdderRebalanceListener implements PollIntervalAwareConsumerRebalanc partitions.forEach(tp -> { Integer partition = tp.partition(); - Long newOffset = consumer.position(tp); + Long offset = consumer.position(tp); log.info( "{} - removing partition: {}, offset of next message {})", id, partition, - newOffset); + offset); if (commitsEnabled) { - repository.save(new StateDocument(partition, handler.removePartition(partition), newOffset)); + repository.save(new StateDocument(partition, handler.removePartition(partition), offset)); } else {