Überprüfung des ``BackOffState``-Zustands berücksichtigt immer den Offset

diff --git a/src/main/java/de/juplo/kafka/BackOffState.java b/src/main/java/de/juplo/kafka/BackOffState.java
index 0015ebec586f8ec91de15d55b1de75699f47c0e9..4981bb6c45fe314529d23eb25c0bd2f6cd93b858 100644 (file)
--- a/src/main/java/de/juplo/kafka/BackOffState.java
+++ b/src/main/java/de/juplo/kafka/BackOffState.java
@@ -90,9 +90,9 @@ class BackOffState
     return this.offset == offset && backOffExecution != null;
   }
 
-  boolean isCompleted()
+  boolean isCompleted(long offset)
   {
-    return timeNextRetryIsDue == null;
+    return this.offset == offset && timeNextRetryIsDue == null;
   }
 
   void markAsCompleted()

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index b48baaa4ebaeaf1c1a43abdd446bce4e02a0281f..9e73b1196443f09432ae7123543c303bf891c4c3 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -155,7 +155,7 @@ public class ExampleConsumer implements ConsumerRebalanceListener, Runnable
                 }
                 else
                 {
-                  if (backOffState[partition].isCompleted())
+                  if (backOffState[partition].isCompleted(record.offset()))
                   {
                     log.warn("{} - Ignoring retryable error: {}", id, e.toString());
                   }

diff --git a/src/test/java/de/juplo/kafka/BackOffStateTest.java b/src/test/java/de/juplo/kafka/BackOffStateTest.java
index 624b354b5ce3acc04cd7d26a778aeb8964471506..338631c156eb7df599f31eea78972e617c878ead 100644 (file)
--- a/src/test/java/de/juplo/kafka/BackOffStateTest.java
+++ b/src/test/java/de/juplo/kafka/BackOffStateTest.java
@@ -108,7 +108,7 @@ class BackOffStateTest
     // WHEN
 
     // THEN
-    assertThat(backOffState.isCompleted()).isTrue();
+    assertThat(backOffState.isCompleted(OFFSET)).isTrue();
   }
 
 
@@ -163,7 +163,7 @@ class BackOffStateTest
     backOffState.isWaitingForNextRetry();
 
     // THEN
-    assertThat(backOffState.isCompleted()).isFalse();
+    assertThat(backOffState.isCompleted(OFFSET)).isFalse();
   }
 
   @Test
@@ -211,7 +211,7 @@ class BackOffStateTest
     backOffState.isWaitingForNextRetry();
 
     // THEN
-    assertThat(backOffState.isCompleted()).isFalse();
+    assertThat(backOffState.isCompleted(OFFSET)).isFalse();
   }
 
   @Test
@@ -259,7 +259,7 @@ class BackOffStateTest
     backOffState.isWaitingForNextRetry();
 
     // THEN
-    assertThat(backOffState.isCompleted()).isTrue();
+    assertThat(backOffState.isCompleted(OFFSET)).isTrue();
   }
 
   @Test