X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;fp=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessConsumer.java;h=a93ae2cf415fd9e8ab0bac84d4902082ab2fc177;hb=6ded138c6b8139da2cdc13f2380b5f5a4e51cc4e;hp=3d154c2cbd066e430d7f6f2c8ab49f41a3c9d0eb;hpb=ce840f48340d55613291fca468bf10b834c473db;p=demos%2Fkafka%2Ftraining

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index 3d154c2..a93ae2c 100644
--- 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);
     });
   }