Update processor Id and record type handlers

Author: loicgreffier

dsl/src/main/java/com/michelin/kafka/error/handling/dsl/handler/ProcessorIdProcessingHandler.java

@@ -35,6 +35,6 @@ record != null ? record.value() : null,
 
     @Override
     public void configure(Map<String, ?> map) {
-
+        // Do nothing
     }
 }

dsl/src/main/java/com/michelin/kafka/error/handling/dsl/handler/RecordTypeProcessingHandler.java

@@ -5,20 +5,33 @@
 import org.apache.kafka.streams.errors.ErrorHandlerContext;
 import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
 import org.apache.kafka.streams.processor.api.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class RecordTypeProcessingHandler implements ProcessingExceptionHandler {
+    private static final Logger log = LoggerFactory.getLogger(RecordTypeProcessingHandler.class);
+
     @Override
     public ProcessingHandlerResponse handle(ErrorHandlerContext context, Record<?, ?> record, Exception exception) {
-        if (record.value() instanceof DeliveryBooked deliveryBooked) {
+        log.warn(
+            "Exception caught for processorNodeId = {}, topic = {}, partition = {}, offset = {}, key = {}, value = {}",
+            context.processorNodeId(),
+            context.topic(),
+            context.partition(),
+            context.offset(),
+            record != null ? record.key() : null,
+            record != null ? record.value() : null,
+            exception);
+
+        if (record != null && record.value() instanceof DeliveryBooked deliveryBooked) {
             return deliveryBooked.getNumberOfTires() == null ? ProcessingHandlerResponse.CONTINUE : ProcessingHandlerResponse.FAIL;
         }
 
         return ProcessingHandlerResponse.CONTINUE;
-
     }
 
     @Override
     public void configure(Map<String, ?> map) {
-
+        // Do nothing
     }
 }

processor-api/src/main/java/com/michelin/kafka/error/handling/papi/handler/ProcessorIdProcessingHandler.java

@@ -35,6 +35,6 @@ record != null ? record.value() : null,
 
     @Override
     public void configure(Map<String, ?> map) {
-
+        // Do nothing
     }
 }

processor-api/src/main/java/com/michelin/kafka/error/handling/papi/handler/RecordTypeProcessingHandler.java

@@ -5,20 +5,33 @@
 import org.apache.kafka.streams.errors.ErrorHandlerContext;
 import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
 import org.apache.kafka.streams.processor.api.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class RecordTypeProcessingHandler implements ProcessingExceptionHandler {
+    private static final Logger log = LoggerFactory.getLogger(RecordTypeProcessingHandler.class);
+
     @Override
     public ProcessingHandlerResponse handle(ErrorHandlerContext context, Record<?, ?> record, Exception exception) {
-        if (record.value() instanceof DeliveryBooked deliveryBooked) {
+        log.warn(
+            "Exception caught for processorNodeId = {}, topic = {}, partition = {}, offset = {}, key = {}, value = {}",
+            context.processorNodeId(),
+            context.topic(),
+            context.partition(),
+            context.offset(),
+            record != null ? record.key() : null,
+            record != null ? record.value() : null,
+            exception);
+
+        if (record != null && record.value() instanceof DeliveryBooked deliveryBooked) {
             return deliveryBooked.getNumberOfTires() == null ? ProcessingHandlerResponse.CONTINUE : ProcessingHandlerResponse.FAIL;
         }
 
         return ProcessingHandlerResponse.CONTINUE;
-
     }
 
     @Override
     public void configure(Map<String, ?> map) {
-
+        // Do nothing
     }
 }