Enhance unit testing

Author: julienbrunet

retryable-consumer-core/src/test/java/com/michelin/kafka/test/unit/DefaultErrorProcessorTest.java

@@ -19,6 +19,7 @@
 package com.michelin.kafka.test.unit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
 
@@ -31,6 +32,10 @@
 import java.io.ObjectInputStream;
 import java.nio.ByteBuffer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.RecordDeserializationException;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.record.TimestampType;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
@@ -142,4 +147,31 @@ void shouldHandleErrorWithNullThrowable() {
         assertEquals(record.key(), capturedErrorModel.getKey());
         assertEquals(record.value(), capturedErrorModel.getValue());
     }
+
+    @Test
+    void shouldHandleErrorWithRecordDeserializationException() throws IOException {
+        // Given
+        ConsumerRecord<String, String> record = new ConsumerRecord<>("testTopic", 12, 13458L, "key", "value");
+        RecordDeserializationException rde =
+                new RecordDeserializationException(
+                        RecordDeserializationException.DeserializationExceptionOrigin.KEY,
+                        new TopicPartition(record.topic(),record.partition()),
+                        record.offset(),
+                        1764603801,
+                        TimestampType.CREATE_TIME,
+                        DefaultErrorProcessor.toByteBuffer("MyKey"),
+                        DefaultErrorProcessor.toByteBuffer("MyValue"),
+                        null,
+                        "Record deserialization error",
+                        null);
+
+        // When
+        errorProcessor.processError(rde, record, 32L);
+        verify(mockDeadLetterProducer, times(1)).send(keyCaptor.capture(), valueCaptor.capture());
+
+        GenericErrorModel capturedErrorModel = valueCaptor.getValue();
+
+        assertEquals("Record deserialization error", capturedErrorModel.getCause());
+        assertEquals(record.topic(), capturedErrorModel.getTopic());
+    }
 }

retryable-consumer-core/src/test/java/com/michelin/kafka/test/unit/RetryableConsumerErrorHandlerTest.java

@@ -160,7 +160,7 @@ void shouldHandleCustomErrorProcessor() {
         ConsumerRecord<String, String> record = new ConsumerRecord<>(topic, partition, offset, key, value);
 
         // When
-        errorHandlerWithCustomErrorProcessor.handleError(exception, record);
+        errorHandlerWithCustomErrorProcessor.handleError(exception, record, 2L);
 
         // Then
         verify(mockDeadLetterProducer, times(0)).send(any(), any()); // Verify the default error processor is not used
@@ -173,6 +173,7 @@ void shouldHandleCustomErrorProcessor() {
         assertEquals(record.topic(), recordCaptor.getValue().topic());
         assertEquals(record.offset(), recordCaptor.getValue().offset());
         assertEquals(record.partition(), recordCaptor.getValue().partition());
+        assertEquals(2L, retryCountCaptor.getValue());
     }
 
     public static class SerializableObject implements Serializable {