juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/PartitionStatistics.java

   1 package de.juplo.kafka;
   2
   3 import lombok.EqualsAndHashCode;
   4 import lombok.Getter;
   5 import lombok.RequiredArgsConstructor;
   6 import org.apache.kafka.common.TopicPartition;
   7
   8 import java.util.Collection;
   9 import java.util.HashMap;
  10 import java.util.Map;
  11
  12
  13 @RequiredArgsConstructor
  14 @Getter
  15 @EqualsAndHashCode(of = { "partition" })
  16 public class PartitionStatistics
  17 {
  18   private final TopicPartition partition;
  19   private final Map<String, KeyCounter> statistics = new HashMap<>();
  20
  21
  22   public KeyCounter addKey(String key)
  23   {
  24     KeyCounter counter = new KeyCounter(key);
  25
  26     counter.increment();
  27     statistics.put(key, counter);
  28
  29     return counter;
  30   }
  31
  32
  33   public long increment(String key)
  34   {
  35     KeyCounter counter = statistics.get(key);
  36
  37     if (counter == null)
  38     {
  39       counter = new KeyCounter(key);
  40       statistics.put(key, counter);
  41     }
  42
  43     return counter.increment();
  44   }
  45
  46   public Collection<KeyCounter> getStatistics()
  47   {
  48     return statistics.values();
  49   }
  50
  51   @Override
  52   public String toString()
  53   {
  54     return partition.toString();
  55   }
  56 }