Endopint zum abfragen der Offsets der zugeordneten Partitionen

[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 803c76e..35e836c 100644 (file)
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@ -179,6 +179,11 @@ public class EndlessConsumer implements Runnable
     future = executor.submit(this);
   }
 
+  public Optional<Long> getOffset(Integer partition)
+  {
+    return Optional.ofNullable(offsets.get(partition));
+  }
+
   public synchronized void stop() throws ExecutionException, InterruptedException
   {
     boolean stateChanged = running.compareAndSet(true, false);