Add support for batch error strategy

micronaut-projects · Oct 4, 2023 · 0a92e19 · 0a92e19
1 parent f2733e8
commit 0a92e19
Show file tree

Hide file tree

Showing 5 changed files with 464 additions and 251 deletions.
diff --git a/kafka/src/main/java/io/micronaut/configuration/kafka/exceptions/KafkaListenerException.java b/kafka/src/main/java/io/micronaut/configuration/kafka/exceptions/KafkaListenerException.java
@@ -15,9 +15,11 @@
  */
 package io.micronaut.configuration.kafka.exceptions;
 
+import io.micronaut.core.annotation.Nullable;
 import io.micronaut.messaging.exceptions.MessageListenerException;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import java.util.Optional;
 
 /**
@@ -31,6 +33,7 @@ public class KafkaListenerException extends MessageListenerException {
 
     private final Object listener;
     private final Consumer<?, ?> kafkaConsumer;
+    private final ConsumerRecords<?, ?> consumerRecords;
     private final ConsumerRecord<?, ?> consumerRecord;
 
     /**
@@ -42,10 +45,7 @@ public class KafkaListenerException extends MessageListenerException {
      * @param consumerRecord The consumer record
      */
     public KafkaListenerException(String message, Object listener, Consumer<?, ?> kafkaConsumer, ConsumerRecord<?, ?> consumerRecord) {
-        super(message);
-        this.listener = listener;
-        this.kafkaConsumer = kafkaConsumer;
-        this.consumerRecord = consumerRecord;
+        this(message, null, listener, kafkaConsumer, consumerRecord);
     }
 
     /**
@@ -58,12 +58,9 @@ public KafkaListenerException(String message, Object listener, Consumer<?, ?> ka
      * @param consumerRecord The consumer record
      */
     public KafkaListenerException(String message, Throwable cause, Object listener, Consumer<?, ?> kafkaConsumer, ConsumerRecord<?, ?> consumerRecord) {
-        super(message, cause);
-        this.listener = listener;
-        this.kafkaConsumer = kafkaConsumer;
-        this.consumerRecord = consumerRecord;
+        this(message, cause, listener, kafkaConsumer, null, consumerRecord);
     }
-    
+
     /**
      * Creates a new exception.
      *
@@ -73,9 +70,31 @@ public KafkaListenerException(String message, Throwable cause, Object listener,
      * @param consumerRecord The consumer record
      */
     public KafkaListenerException(Throwable cause, Object listener, Consumer<?, ?> kafkaConsumer, ConsumerRecord<?, ?> consumerRecord) {
-        super(cause.getMessage(), cause);
+        this(cause.getMessage(), cause, listener, kafkaConsumer, consumerRecord);
+    }
+
+    /**
+     * Creates a new exception.
+     *
+     * @param message The message
+     * @param cause The cause
+     * @param listener The listener
+     * @param kafkaConsumer The consumer
+     * @param consumerRecords The batch of consumer records
+     * @param consumerRecord The consumer record
+     */
+    public KafkaListenerException(
+        String message,
+        Throwable cause,
+        Object listener,
+        Consumer<?, ?> kafkaConsumer,
+        @Nullable ConsumerRecords<?, ?> consumerRecords,
+        @Nullable ConsumerRecord<?, ?> consumerRecord
+    ) {
+        super(message, cause);
         this.listener = listener;
         this.kafkaConsumer = kafkaConsumer;
+        this.consumerRecords = consumerRecords;
         this.consumerRecord = consumerRecord;
     }
 
@@ -99,4 +118,12 @@ public Object getKafkaListener() {
     public Optional<ConsumerRecord<?, ?>> getConsumerRecord() {
         return Optional.ofNullable(consumerRecord);
     }
+
+    /**
+     * @return The batch of consumer records that was being processed that caused the error
+     * @since 5.3
+     */
+    public Optional<ConsumerRecords<?, ?>> getConsumerRecords() {
+        return Optional.ofNullable(consumerRecords);
+    }
 }