feat(kafka): support max. number of retries in RetryProcessingErrorMLS

Author: mibutec

sda-commons-server-kafka/src/main/java/org/sdase/commons/server/kafka/consumer/strategies/retryprocessingerror/RetryCounter.java

@@ -0,0 +1,65 @@
+package org.sdase.commons.server.kafka.consumer.strategies.retryprocessingerror;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+/**
+ * Helper class counting the number of executions for a given record.
+ *
+ * <p>Internally it increases a counter for the current offset of the record as long as the offset
+ * remains the same. If the offset changes, the counter is reset to 0.
+ */
+public class RetryCounter {
+  private final Map<Integer, OffsetCounter> offsetCountersByPartition =
+      Collections.synchronizedMap(new HashMap<>());
+
+  private final long maxRetryCount;
+
+  public RetryCounter(long maxRetryCount) {
+    this.maxRetryCount = maxRetryCount;
+  }
+
+  public void incErrorCount(ConsumerRecord<?, ?> consumerRecord) {
+    getOffsetCounterInternal(consumerRecord).inc();
+  }
+
+  public boolean isMaxRetryCountReached(ConsumerRecord<?, ?> consumerRecord) {
+    long counter = getOffsetCounter(consumerRecord);
+    return counter > maxRetryCount;
+  }
+
+  public long getOffsetCounter(ConsumerRecord<?, ?> consumerRecord) {
+    return getOffsetCounterInternal(consumerRecord).count;
+  }
+
+  public long getMaxRetryCount() {
+    return maxRetryCount;
+  }
+
+  private OffsetCounter getOffsetCounterInternal(ConsumerRecord<?, ?> consumerRecord) {
+    OffsetCounter counter =
+        offsetCountersByPartition.computeIfAbsent(
+            consumerRecord.partition(), k -> new OffsetCounter(0));
+    if (counter.offset != consumerRecord.offset()) {
+      counter.offset = consumerRecord.offset();
+      counter.count = 0;
+    }
+    return counter;
+  }
+
+  private static class OffsetCounter {
+    private long offset = 0;
+    private long count = 0;
+
+    public OffsetCounter(long offset) {
+      this.offset = offset;
+      this.count = 0;
+    }
+
+    public long inc() {
+      return ++count;
+    }
+  }
+}

sda-commons-server-kafka/src/main/java/org/sdase/commons/server/kafka/consumer/strategies/retryprocessingerror/RetryProcessingErrorMLS.java

@@ -21,18 +21,57 @@
 
 /**
  * {@link MessageListenerStrategy} commits records for each partition. In case of processing errors
- * an error handler can decide if processing should be retried or stopped.
+ * the message will be retried for a configured amount of times.
+ *
+ * <p>After each error an error handler can decide if processing should be retried or stopped.
+ *
+ * <p>If the retry count exceeds:
+ *
+ * <ul>
+ *   <li>The failed message handling will be logged as ERROR
+ *   <li>a retryLimitExceededErrorHandler will be called allowing to do more error handling (i.e.
+ *       write to DLT)
+ *   <li>the offset will get increased, so the message handling gets ignored.
+ * </ul>
  */
 public class RetryProcessingErrorMLS<K, V> extends MessageListenerStrategy<K, V> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RetryProcessingErrorMLS.class);
   private final MessageHandler<K, V> handler;
   private final ErrorHandler<K, V> errorHandler;
+  private final ErrorHandler<K, V> retryLimitExceededErrorHandler;
   private String consumerName;
+  private final RetryCounter retryCounter;
 
+  /**
+   * Creates a new instance of {@link RetryProcessingErrorMLS} retrying the message on error
+   * infinite times.
+   *
+   * @param handler the message handler
+   * @param errorHandler the error handler called after each error
+   */
   public RetryProcessingErrorMLS(MessageHandler<K, V> handler, ErrorHandler<K, V> errorHandler) {
+    this(handler, errorHandler, Long.MAX_VALUE, null);
+  }
+
+  /**
+   * Creates a new instance of {@link RetryProcessingErrorMLS} retrying the message on error for a
+   * given amount of times
+   *
+   * @param handler the message handler
+   * @param errorHandler the error handler called after each error
+   * @param maxRetryCount the maximum number of retries
+   * @param retryLimitExceededErrorHandler the error handler called if the retry limit is exceeded
+   */
+  public RetryProcessingErrorMLS(
+      MessageHandler<K, V> handler,
+      ErrorHandler<K, V> errorHandler,
+      long maxRetryCount,
+      ErrorHandler<K, V> retryLimitExceededErrorHandler) {
     this.handler = handler;
     this.errorHandler = errorHandler;
+    this.retryLimitExceededErrorHandler = retryLimitExceededErrorHandler;
+    this.retryCounter = new RetryCounter(maxRetryCount);
   }
 
   @Override
@@ -57,9 +96,7 @@ private void processRecordsByPartition(
         try {
           Instant timerStart = Instant.now();
           handler.handle(consumerRecord);
-          // mark last successful processed record for commit
-          lastCommitOffset = new OffsetAndMetadata(consumerRecord.offset() + 1);
-          addOffsetToCommitOnClose(consumerRecord);
+          lastCommitOffset = markConsumerRecordProcessed(consumerRecord);
 
           Instant timerEnd = Instant.now();
           if (LOGGER.isTraceEnabled()) {
@@ -69,23 +106,28 @@ private void processRecordsByPartition(
                 consumerName,
                 consumerRecord.topic());
           }
-
         } catch (RuntimeException e) {
-          LOGGER.error(
-              "Error while handling record {} in message handler {}",
-              consumerRecord.key(),
-              handler.getClass(),
-              e);
-          boolean shouldContinue = errorHandler.handleError(consumerRecord, e, consumer);
-          if (!shouldContinue) {
-            throw new StopListenerException(e);
+          retryCounter.incErrorCount(consumerRecord);
+          if (retryCounter.isMaxRetryCountReached(consumerRecord)) {
+            LOGGER.error(
+                "Error while handling record {} in message handler {}, no more retries",
+                consumerRecord.key(),
+                handler.getClass(),
+                e);
+
+            callErrorHandler(retryLimitExceededErrorHandler, consumerRecord, e, consumer);
+            lastCommitOffset = markConsumerRecordProcessed(consumerRecord);
           } else {
             LOGGER.warn(
-                "Error while handling record {} in message handler {}, will be retried",
+                "Error while handling record {} in message handler {}, will be retried ({} / {})...",
                 consumerRecord.key(),
                 handler.getClass(),
+                retryCounter.getOffsetCounter(consumerRecord),
+                retryCounter.getMaxRetryCount(),
                 e);
 
+            callErrorHandler(errorHandler, consumerRecord, e, consumer);
+
             // seek to the current offset of the failing record for retry
             consumer.seek(partition, consumerRecord.offset());
             break;
@@ -94,6 +136,7 @@ private void processRecordsByPartition(
       }
     }
     if (lastCommitOffset != null) {
+
       consumer.commitSync(Collections.singletonMap(partition, lastCommitOffset));
     }
   }
@@ -105,4 +148,22 @@ public void verifyConsumerConfig(Map<String, String> config) {
           "The strategy should commit explicitly by partition but property 'enable.auto.commit' in consumer config is set to 'true'");
     }
   }
+
+  private void callErrorHandler(
+      ErrorHandler<K, V> errorHandler,
+      ConsumerRecord<K, V> consumerRecord,
+      RuntimeException e,
+      KafkaConsumer<K, V> consumer) {
+    if (errorHandler != null) {
+      boolean shouldContinue = errorHandler.handleError(consumerRecord, e, consumer);
+      if (!shouldContinue) {
+        throw new StopListenerException(e);
+      }
+    }
+  }
+
+  private OffsetAndMetadata markConsumerRecordProcessed(ConsumerRecord<K, V> consumerRecord) {
+    addOffsetToCommitOnClose(consumerRecord);
+    return new OffsetAndMetadata(consumerRecord.offset() + 1);
+  }
 }