From c6c3d6879d0fcf0a8937733f23ff1c487fb4426c Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sat, 9 Apr 2022 18:00:48 +0200
Subject: [PATCH] =?utf8?q?Gesehene=20Schl=C3=BCssel=20sollten=20als=20long?=
 =?utf8?q?=20gez=C3=A4hlt=20werden?=
MIME-Version: 1.0
Content-Type: text/plain; charset=utf8
Content-Transfer-Encoding: 8bit

---
 src/main/java/de/juplo/kafka/DriverController.java |  2 +-
 src/main/java/de/juplo/kafka/EndlessConsumer.java  | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/src/main/java/de/juplo/kafka/DriverController.java b/src/main/java/de/juplo/kafka/DriverController.java
index a504842..93e2856 100644
--- a/src/main/java/de/juplo/kafka/DriverController.java
+++ b/src/main/java/de/juplo/kafka/DriverController.java
@@ -30,7 +30,7 @@ public class DriverController
 
 
   @GetMapping("seen")
-  public Map<Integer, Map<String, Integer>> seen()
+  public Map<Integer, Map<String, Long>> seen()
   {
     return consumer.getSeen();
   }
diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 14a875b..2e79009 100644
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -33,7 +33,7 @@ public class EndlessConsumer implements Runnable
   private KafkaConsumer<String, String> consumer = null;
   private Future<?> future = null;
 
-  private final Map<Integer, Map<String, Integer>> seen = new HashMap<>();
+  private final Map<Integer, Map<String, Long>> seen = new HashMap<>();
 
 
   public EndlessConsumer(
@@ -77,7 +77,7 @@ public class EndlessConsumer implements Runnable
           partitions.forEach(tp ->
           {
             log.info("{} - removing partition: {}", id, tp);
-            Map<String, Integer> removed = seen.remove(tp.partition());
+            Map<String, Long> removed = seen.remove(tp.partition());
             for (String key : removed.keySet())
             {
               log.info(
@@ -123,12 +123,12 @@ public class EndlessConsumer implements Runnable
 
           Integer partition = record.partition();
           String key = record.key() == null ? "NULL" : record.key();
-          Map<String, Integer> byKey = seen.get(partition);
+          Map<String, Long> byKey = seen.get(partition);
 
           if (!byKey.containsKey(key))
-            byKey.put(key, 0);
+            byKey.put(key, 0l);
 
-          int seenByKey = byKey.get(key);
+          long seenByKey = byKey.get(key);
           seenByKey++;
           byKey.put(key, seenByKey);
         }
@@ -151,7 +151,7 @@ public class EndlessConsumer implements Runnable
     }
   }
 
-  public Map<Integer, Map<String, Integer>> getSeen()
+  public Map<Integer, Map<String, Long>> getSeen()
   {
     return seen;
   }
-- 
2.20.1