From 3c1d3fa68df685146bdef7cc2e396e55fa0933dc Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Thu, 7 Apr 2022 00:00:57 +0200
Subject: [PATCH] Offset-Position in der MongoDB wird vor jedem `poll()`
 aktualisiert

---
 src/main/java/de/juplo/kafka/EndlessConsumer.java | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 69c8294..b152310 100644
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -135,6 +135,8 @@ public class EndlessConsumer implements Runnable
           String key = record.key() == null ? "NULL" : record.key();
           seen.get(partition).increment(key);
         }
+
+        seen.forEach((tp, statistics) -> repository.save(new StatisticsDocument(statistics, consumer.position(tp))));
       }
     }
     catch(WakeupException e)
-- 
2.20.1