X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationRebalanceListener.java;h=32e14e8159774f1421cb65a045939ac27e7e0e28;hb=890fd85c334a078610701bd6e571d133df69473f;hp=7256732b81f9c1649021ca19e54171ca0d53307a;hpb=e5c029af3690ae8ebed729af4f06296cef13fa3c;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index 7256732..32e14e8 100644
--- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@ -7,8 +7,7 @@ import org.apache.kafka.common.TopicPartition;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
-import java.util.Collection;
-import java.util.Map;
+import java.util.*;
 
 
 @RequiredArgsConstructor
@@ -16,11 +15,14 @@ import java.util.Map;
 public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRebalanceListener
 {
   private final ApplicationRecordHandler recordHandler;
+  private final AdderResults adderResults;
   private final StateRepository stateRepository;
   private final String id;
   private final Clock clock;
   private final Duration commitInterval;
 
+  private final Set<Integer> partitions = new HashSet<>();
+
   private Instant lastCommit = Instant.EPOCH;
 
   @Override
@@ -30,11 +32,13 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
     {
       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));
       recordHandler.addPartition(partition, document.state);
+      adderResults.addPartition(partition, document.results);
     });
   }
 
@@ -45,17 +49,19 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
     {
       Integer partition = tp.partition();
       log.info("{} - removing partition: {}", id, partition);
-      Map<String, Long> removed = recordHandler.removePartition(partition);
-      for (String key : removed.keySet())
+      this.partitions.remove(partition);
+      Map<String, AdderResult> state = recordHandler.removePartition(partition);
+      for (String key : state.keySet())
       {
         log.info(
             "{} - Seen {} messages for partition={}|key={}",
             id,
-            removed.get(key),
+            state.get(key),
             partition,
             key);
       }
-      stateRepository.save(new StateDocument(partition, removed));
+      Map<String, List<AdderResult>> results = adderResults.removePartition(partition);
+      stateRepository.save(new StateDocument(partition, state, results));
     });
   }
 
@@ -66,10 +72,11 @@ public class ApplicationRebalanceListener implements PollIntervalAwareConsumerRe
     if (lastCommit.plus(commitInterval).isBefore(clock.instant()))
     {
       log.debug("Storing data, last commit: {}", lastCommit);
-      recordHandler.getState().forEach((partiton, adder) -> stateRepository.save(
+      partitions.forEach(partition -> stateRepository.save(
           new StateDocument(
-              partiton,
-              adder.getState())));
+              partition,
+              recordHandler.getState(partition).getState(),
+              adderResults.getState(partition))));
       lastCommit = clock.instant();
     }
   }