`auto.offset.reset` konfigurierbar gemacht

diff --git a/src/main/java/de/juplo/kafka/Application.java b/src/main/java/de/juplo/kafka/Application.java
index 85d0e079e50355e09378bf7b9e47150dca5752bd..dd4b20a02dd4d8eae5160b49e476817ca80874e7 100644 (file)
--- a/src/main/java/de/juplo/kafka/Application.java
+++ b/src/main/java/de/juplo/kafka/Application.java
@@ -32,7 +32,8 @@ public class Application
             properties.getBootstrapServer(),
             properties.getGroupId(),
             properties.getClientId(),
-            properties.getTopic());
+            properties.getTopic(),
+            properties.getAutoOffsetReset());
 
     consumer.start();
 

diff --git a/src/main/java/de/juplo/kafka/ApplicationProperties.java b/src/main/java/de/juplo/kafka/ApplicationProperties.java
index fdbb2bdff831206e98aef603fd9e7d4b764e5f7e..dab33801def947153ed43afc087ff2ce0a2cfadc 100644 (file)
--- a/src/main/java/de/juplo/kafka/ApplicationProperties.java
+++ b/src/main/java/de/juplo/kafka/ApplicationProperties.java
@@ -14,4 +14,5 @@ public class ApplicationProperties
   private String groupId;
   private String clientId;
   private String topic;
+  private String autoOffsetReset;
 }

diff --git a/src/main/java/de/juplo/kafka/EndlessConsumer.java b/src/main/java/de/juplo/kafka/EndlessConsumer.java
index da2f8f0c29653a2420c1769c0db75cbc6db872c3..b3dd446d84ddbc7fddd44d516cbd921eff1ece7e 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());
 

diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
index 763880df46d76397e1798be5308d55fec3ae35bb..db378220e49bac448462a6faaab48d27193ff235 100644 (file)
--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -3,6 +3,7 @@ consumer:
   group-id: my-consumer
   client-id: peter
   topic: test
+  auto-offset-reset: earliest
 management:
   endpoints:
     web: