From: Kai Moritz Date: Sun, 14 Aug 2022 17:14:47 +0000 (+0200) Subject: fix: In `onPartitionsAssigned()` wurde der Kafka-Offset ausgegeben X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=824f55bc813acebd4ea67ccf7f22518a31351076;hp=-c;p=demos%2Fkafka%2Ftraining fix: In `onPartitionsAssigned()` wurde der Kafka-Offset ausgegeben --- 824f55bc813acebd4ea67ccf7f22518a31351076 diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java index 2fccb4f..59b420a 100644 --- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java +++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java @@ -32,12 +32,11 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe partitions.forEach(tp -> { Integer partition = tp.partition(); - Long offset = consumer.position(tp); - log.info("{} - adding partition: {}, offset={}", id, partition, offset); StateDocument document = stateRepository .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,14 +53,14 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe 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); Map removed = recordHandler.removePartition(partition); - stateRepository.save(new StateDocument(partition, removed, consumer.position(tp))); + stateRepository.save(new StateDocument(partition, removed, offset)); }); }