`auto.offset.reset` konfigurierbar gemacht

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

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index da2f8f0..b3dd446 100644 (file)
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -25,6 +25,7 @@ public class EndlessConsumer implements Runnable
   private final String groupId;
   private final String id;
   private final String topic;
+  private final String autoOffsetReset;
 
   private AtomicBoolean running = new AtomicBoolean();
   private long consumed = 0;
@@ -36,13 +37,15 @@ public class EndlessConsumer implements Runnable
       String bootstrapServer,
       String groupId,
       String clientId,
-      String topic)
+      String topic,
+      String autoOffsetReset)
   {
     this.executor = executor;
     this.bootstrapServer = bootstrapServer;
     this.groupId = groupId;
     this.id = clientId;
     this.topic = topic;
+    this.autoOffsetReset = autoOffsetReset;
   }
 
   @Override
@@ -54,7 +57,7 @@ public class EndlessConsumer implements Runnable
       props.put("bootstrap.servers", bootstrapServer);
       props.put("group.id", groupId);
       props.put("client.id", id);
-      props.put("auto.offset.reset", "earliest");
+      props.put("auto.offset.reset", autoOffsetReset);
       props.put("key.deserializer", StringDeserializer.class.getName());
       props.put("value.deserializer", StringDeserializer.class.getName());