Die Key-Statistiken werden in einer MongoDB gespeichert

[demos/kafka/training] / src / main / java / de / juplo / kafka / PartitionStatistics.java

diff --git a/src/main/java/de/juplo/kafka/PartitionStatistics.java b/src/main/java/de/juplo/kafka/PartitionStatistics.java
index e47a9f9..0e31945 100644 (file)
--- a/src/main/java/de/juplo/kafka/PartitionStatistics.java
+++ b/src/main/java/de/juplo/kafka/PartitionStatistics.java
@@ -2,7 +2,6 @@ package de.juplo.kafka;
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
-import lombok.RequiredArgsConstructor;
 import org.apache.kafka.common.TopicPartition;
 
 import java.util.Collection;
@@ -10,15 +9,35 @@ import java.util.HashMap;
 import java.util.Map;
 
 
-@RequiredArgsConstructor
 @Getter
 @EqualsAndHashCode(of = { "partition" })
 public class PartitionStatistics
 {
+  private String id;
   private final TopicPartition partition;
   private final Map<String, KeyCounter> statistics = new HashMap<>();
 
 
+  public PartitionStatistics(TopicPartition partition)
+  {
+    this.partition = partition;
+  }
+
+  public PartitionStatistics(StatisticsDocument document)
+  {
+    this.partition = new TopicPartition(document.topic, document.partition);
+    document
+        .statistics
+        .entrySet()
+        .forEach(entry ->
+        {
+          this.statistics.put(
+              entry.getKey(),
+              new KeyCounter(entry.getKey(), entry.getValue()));
+        });
+  }
+
+
   public KeyCounter addKey(String key)
   {
     KeyCounter counter = new KeyCounter(key);