FIX:double-mark-as-successful

diff --git a/src/main/java/de/juplo/kafka/BackOffState.java b/src/main/java/de/juplo/kafka/BackOffState.java
index 556b3b1..b8a226b 100644 (file)
--- a/src/main/java/de/juplo/kafka/BackOffState.java
+++ b/src/main/java/de/juplo/kafka/BackOffState.java
@@ -95,9 +95,9 @@ class BackOffState
     return Duration.between(startTime, timeNextRetryIsDue);
   }
 
-  void markRetryAsSuccessful()
+  void markRetryAsSuccessful(long offset)
   {
-    if (inProgress)
+    if (inProgress && this.offset == offset)
     {
       Instant now = clock.instant();
       log.info(

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 2ef4b0a..51e69ff 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -185,7 +185,7 @@ public class ExampleConsumer implements ConsumerRebalanceListener, Runnable
                 log.warn("{} - Ignoring non-retryable error!", id, e);
               }
 
-              backOffState[topicPartition.partition()].markRetryAsSuccessful();
+              backOffState[topicPartition.partition()].markRetryAsSuccessful(record.offset());
             }
           }
         }