From: Kai Moritz <kai@juplo.de>
Date: Wed, 6 Apr 2022 22:00:57 +0000 (+0200)
Subject: Offset-Position in der MongoDB wird vor jedem `poll()` aktualisiert
X-Git-Tag: wip-DEPRECATED~14
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=3c1d3fa68df685146bdef7cc2e396e55fa0933dc;p=demos%2Fkafka%2Ftraining

Offset-Position in der MongoDB wird vor jedem `poll()` aktualisiert
---

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)