Avoid committing consumer after error

seedstack · Sep 30, 2020 · 3a7afcc · 3a7afcc
1 parent 14c25c9
commit 3a7afcc
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 15 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,3 +1,7 @@
+# Version 2.0.2 (2020-09-30)
+
+* [fix] Avoid wrongly committing Kafka consumer after an error, which consumed the message without any chance to process it.
+
 # Version 2.0.1 (2020-08-05)
 
 * [chg] Updated for seed 3.10+

diff --git a/pom.xml b/pom.xml
@@ -19,7 +19,7 @@
 
     <groupId>org.seedstack.addons.kafka</groupId>
     <artifactId>kafka</artifactId>
-    <version>2.0.1-SNAPSHOT</version>
+    <version>2.0.2-SNAPSHOT</version>
 
     <properties>
         <seed.version>3.10.0</seed.version>

diff --git a/src/main/java/org/seedstack/kafka/internal/ConsumerListenerHandler.java b/src/main/java/org/seedstack/kafka/internal/ConsumerListenerHandler.java
@@ -7,18 +7,7 @@
  */
 package org.seedstack.kafka.internal;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 import com.google.inject.Injector;
-import java.time.Duration;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import javax.inject.Inject;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -31,6 +20,18 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.inject.Inject;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
 class ConsumerListenerHandler<K, V> implements Runnable, ListenerHandler {
     private static final Duration POLLING_DELAY = Duration.ofMillis(1000);
     private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerListenerHandler.class);
@@ -60,6 +61,7 @@ public Class<?> getListenerClass() {
     public void run() {
         String consumerName = annotation.value();
         Duration retryDelay = Duration.ofMillis(annotation.retryDelay());
+        boolean errorOccurred = false;
 
         try {
             ConsumerListener<K, V> listener = injector.getInstance(listenerClass);
@@ -81,6 +83,7 @@ public void run() {
                     }
                 } catch (Exception e1) {
                     // Ignore exception processing if we are stopping
+                    errorOccurred = true;
                     if (active.get()) {
                         try {
                             listener.onException(e1);
@@ -96,13 +99,15 @@ public void run() {
         } finally {
             synchronized (this) {
                 if (consumer != null) {
-                    LOGGER.info("Synchronously committing Kafka consumer: {}", consumerName);
-                    consumer.commitSync();
+                    if (!errorOccurred) {
+                        LOGGER.info("Synchronously committing Kafka consumer: {}", consumerName);
+                        consumer.commitSync();
+                    }
                     LOGGER.info("Closing Kafka consumer: {}", consumerName);
                     try {
                         consumer.close();
                     } catch (Exception e) {
-                        LOGGER.warn("Unable to properly close Kafka consumer: {}", e);
+                        LOGGER.warn("Unable to properly close Kafka consumer: {}", consumerName, e);
                     }
                 }
             }