From 6bfbab41a28c8fb6252d328954f5b7129b24ec25 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 21 Aug 2022 18:10:09 +0200 Subject: [PATCH] Log-Meldungen bei Assign/Revoke verbessert und angeglichen --- .../kafka/ApplicationRebalanceListener.java | 36 +++++++++++++++---- 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java index cd9da64..109b205 100644 --- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java +++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java @@ -35,19 +35,33 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe partitions.forEach(tp -> { Integer partition = tp.partition(); + log.info("{} - adding partition: {}", id, partition); this.partitions.add(partition); 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 // Otherwise: Use initial offset, generated by Kafka consumer.seek(tp, document.offset); + log.info( + "{} - Seeking to offset {} for partition {}", + id, + document.offset, + partition); } recordHandler.addPartition(partition, document.state); + for (String user : document.state.keySet()) + { + log.info( + "{} - Restored state for partition={}|user={}: {}", + id, + partition, + user, + document.state.get(user)); + } adderResults.addPartition(partition, document.results); }); } @@ -58,16 +72,26 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe partitions.forEach(tp -> { Integer partition = tp.partition(); + log.info("{} - removing partition: {}", id, partition); this.partitions.remove(partition); Long offset = consumer.position(tp); - log.info( - "{} - removing partition: {}, offset of next message {})", - id, - partition, - offset); if (commitsEnabled) { + log.info( + "{} - Storing {} as offset of next message for partition {}", + id, + offset, + partition); Map state = recordHandler.removePartition(partition); + for (String user : state.keySet()) + { + log.info( + "{} - Saved state for partition={}|user={}: {}", + id, + partition, + user, + state.get(user)); + } Map> results = adderResults.removePartition(partition); stateRepository.save(new StateDocument(partition, state, results, offset)); } -- 2.20.1