Improve record type processing handler

Author: loicgreffier

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

@@ -44,7 +44,7 @@
 import org.apache.kafka.streams.processor.api.ProcessorContext;
 import org.apache.kafka.streams.processor.api.Record;
 
-public class CustomProcessor extends ContextualProcessor<String, String, String, String> {
+public class CustomProcessor extends ContextualProcessor<String, DeliveryBooked, String, String> {
     private static final Gson gson = new Gson();
 
     @Override
@@ -57,22 +57,16 @@ public void init(ProcessorContext<String, String> context) {
     }
 
     @Override
-    public void process(Record<String, String> message) {
-        DeliveryBooked value = parseFromJson(message.value());
-
-        if (value.getNumberOfTires() < 0) {
+    public void process(Record<String, DeliveryBooked> message) {
+        if (message.value().getNumberOfTires() < 0) {
             throw new InvalidDeliveryException("Number of tires cannot be negative");
         }
 
-        if (value.getNumberOfTires() >= 10) {
-            context().forward(message.withValue(parseToJson(value)));
+        if (message.value().getNumberOfTires() >= 10) {
+            context().forward(message.withValue(parseToJson(message.value())));
         }
     }
 
-    private static DeliveryBooked parseFromJson(String value) {
-        return gson.fromJson(value, DeliveryBooked.class);
-    }
-
     private static String parseToJson(DeliveryBooked value) {
         return gson.toJson(value);
     }

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

@@ -22,6 +22,7 @@
 import static org.apache.kafka.streams.StreamsConfig.BOOTSTRAP_SERVERS_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG;
 
+import com.google.gson.Gson;
 import com.michelin.kafka.error.handling.papi.handler.ExceptionTypeProcessingHandler;
 import java.util.Optional;
 import java.util.Properties;
@@ -34,6 +35,8 @@
 
 /** Kafka Streams application. */
 public class KafkaStreamsApp {
+    private static final Gson gson = new Gson();
+
     public static void main(String[] args) {
         Properties properties = new Properties();
         properties.put(APPLICATION_ID_CONFIG, "processing-error-handling-papi-app");
@@ -56,7 +59,12 @@ public static void main(String[] args) {
 
     public static void buildTopology(StreamsBuilder streamsBuilder) {
         streamsBuilder.stream("delivery_booked_topic", Consumed.with(Serdes.String(), Serdes.String()))
+                .mapValues(KafkaStreamsApp::parseFromJson)
                 .process(CustomProcessor::new)
                 .to("filtered_delivery_booked_papi_topic", Produced.with(Serdes.String(), Serdes.String()));
     }
+
+    private static DeliveryBooked parseFromJson(String value) {
+        return gson.fromJson(value, DeliveryBooked.class);
+    }
 }