X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FWordcountRecordHandler.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FWordcountRecordHandler.java;h=bdf4b328c048171f3124e3e24c4454c60166da25;hb=fc682d9890787ef363b3e189f6f880a043f3c541;hp=5981c7d6e9d4c9404c979f2e230927275ed7f742;hpb=818c1eb862247e25abf9f7d91d5a73e3e3789a39;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/WordcountRecordHandler.java b/src/main/java/de/juplo/kafka/WordcountRecordHandler.java
index 5981c7d..bdf4b32 100644
--- a/src/main/java/de/juplo/kafka/WordcountRecordHandler.java
+++ b/src/main/java/de/juplo/kafka/WordcountRecordHandler.java
@@ -22,7 +22,6 @@ public class WordcountRecordHandler implements RecordHandler<String, String>
 
 
   private final PartitionStatisticsRepository repository;
-  private final String id;
   private final String topic;
   private final Clock clock;
   private final Duration commitInterval;
@@ -78,37 +77,14 @@ public class WordcountRecordHandler implements RecordHandler<String, String>
     }
   }
 
-  @Override
-  public void onPartitionAssigned(TopicPartition tp)
+  public void addPartition(Integer partition, Map<String, Map<String, Long>> statistics)
   {
-    Integer partition = tp.partition();
-    Long offset = consumer.position(tp);
-    log.info("{} - adding partition: {}, offset={}", id, partition, offset);
-    StatisticsDocument document =
-        repository
-            .findById(Integer.toString(partition))
-            .orElse(new StatisticsDocument(partition));
-    if (document.offset >= 0)
-    {
-      // Only seek, if a stored offset was found
-      // Otherwise: Use initial offset, generated by Kafka
-      consumer.seek(tp, document.offset);
-    }
-    seen.put(partition, document.statistics);
+    seen.put(partition, statistics);
   }
 
-  @Override
-  public void onPartitionRevoked(TopicPartition tp)
+  public Map<String, Map<String, Long>> removePartition(Integer partition)
   {
-    Integer partition = tp.partition();
-    Long newOffset = consumer.position(tp);
-    log.info(
-        "{} - removing partition: {}, offset of next message {})",
-        id,
-        partition,
-        newOffset);
-    Map<String, Map<String, Long>> removed = seen.remove(partition);
-    repository.save(new StatisticsDocument(partition, removed, consumer.position(tp)));
+    return seen.remove(partition);
   }