Rückbau der Berücksichtigung von `auto.offset.rest`, wenn Offset unbekannt

[demos/kafka/training] / src / main / java / de / juplo / kafka / ApplicationRebalanceListener.java

diff --git a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
index ca6897c..da72555 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
+++ b/src/main/java/de/juplo/kafka/ApplicationRebalanceListener.java
@@ -33,17 +33,12 @@ public class ApplicationRebalanceListener implements RebalanceListener
           stateRepository
               .findById(Integer.toString(partition))
               .orElse(new StateDocument(partition));
-      if (document.offset >= 0)
-      {
-        // Only seek, if a stored offset was found
-        // Otherwise: Use initial offset, generated by Kafka
-        consumer.seek(tp, document.offset);
-        log.info(
-            "{} - Seeking to offset {} for partition {}",
-            id,
-            document.offset,
-            partition);
-      }
+      log.info(
+        "{} - Offset of next unseen message for partition {}: {}",
+        id,
+        partition,
+        document.offset);
+      consumer.seek(tp, document.offset);
       recordHandler.addPartition(partition, document.state);
       for (String user : document.state.keySet())
       {