Wenn kein gespeicherter Offset vorliegt, auto.offset.reset von Kafka nutzen

* Es wird jetzt nur noch dann ein expliziter Seek durchgeführt, wenn eine
  gespeicherte Offset-Position gefunden wurde.
* Andernfalls wird der von Kafka initialisierte Ausgansgs-Offset verwendet.
* Welchen Offset Kafka vorgibt, hängt von `auto.offset.rest` ab!

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 39e7b1b..3afb4c0 100644 (file)
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -75,7 +75,12 @@ public class EndlessConsumer<K, V> implements ConsumerRebalanceListener, Runnabl
           repository
               .findById(Integer.toString(partition))
               .orElse(new StatisticsDocument(partition));
-      consumer.seek(tp, document.offset);
+      if (document.offset >= 0)
+      {
+        // Only seek, if a stored offset was found
+        // Otherwise: Use initial offset, generated by Kafka
+        consumer.seek(tp, document.offset);
+      }
       seen.put(partition, document.statistics);
     });
   }

diff --git a/src/main/java/de/juplo/kafka/StatisticsDocument.java b/src/main/java/de/juplo/kafka/StatisticsDocument.java
index 28264ec..1244f45 100644 (file)
--- a/src/main/java/de/juplo/kafka/StatisticsDocument.java
+++ b/src/main/java/de/juplo/kafka/StatisticsDocument.java
@@ -14,7 +14,7 @@ public class StatisticsDocument
 {
   @Id
   public String id;
-  public long offset;
+  public long offset = -1l;
   public Map<String, Long> statistics;
 
   public StatisticsDocument()