Updating Kafka consumer to avro specific deserializer.

Author: Shubh-arya

README.md

@@ -24,7 +24,7 @@ services.
 - Create new service for GraphQL client using Apollo library
 - Set up performance test for K6 in local and docker env
 - Integration test with Karate and Wiremock
-- 
+- Retry and DLQ set up 
 
 ### To build a docker image run below command:
 ```shell
@@ -42,4 +42,4 @@ Import the project in IntelliJ and run
 or
 ```shell
 mvnw springboot:run
-```
+```

pom.xml

@@ -44,12 +44,12 @@
             <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-devtools</artifactId>
-            <scope>runtime</scope>
-            <optional>true</optional>
-        </dependency>
+<!--        <dependency>-->
+<!--            <groupId>org.springframework.boot</groupId>-->
+<!--            <artifactId>spring-boot-devtools</artifactId>-->
+<!--            <scope>runtime</scope>-->
+<!--            <optional>true</optional>-->
+<!--        </dependency>-->
 
         <dependency>
             <groupId>org.springframework.boot</groupId>
@@ -267,4 +267,4 @@
             <url>https://packages.confluent.io/maven/</url>
         </repository>
     </repositories>
-</project>
+</project>

src/main/java/com/sarya/graphql/service/application/port/in/ProductConsumer.java

@@ -1,7 +1,7 @@
 package com.sarya.graphql.service.application.port.in;
 
+import com.sarya.graphql.service.ProductCreatedEvent;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.avro.generic.GenericRecord;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
@@ -10,7 +10,7 @@
 public class ProductConsumer {
 
   @KafkaListener(topics = {"create-product"}, groupId = "sample-service-group")
-  public void consumeProducts(GenericRecord event) {
+  public void consumeProducts(ProductCreatedEvent event) {
     log.info("message: {}", event);
   }
 }

src/main/java/com/sarya/graphql/service/infrastructure/config/KafkaConsumerConfig.java

@@ -17,11 +17,11 @@
 @Slf4j
 public class KafkaConsumerConfig {
 
-  KafkaProperties kafkaProperties;
-
   @Bean
   @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
-  ConcurrentKafkaListenerContainerFactory<String, ProductCreatedEvent> kafkaListenerContainerFactory() {
+  ConcurrentKafkaListenerContainerFactory<String, ProductCreatedEvent> kafkaListenerContainerFactory(
+      KafkaProperties kafkaProperties
+  ) {
     var consumerFactory = new DefaultKafkaConsumerFactory<String, ProductCreatedEvent>(
         kafkaProperties.buildConsumerProperties()
     );

src/main/resources/application.yml

@@ -14,11 +14,21 @@ spring:
       bootstrap-servers: localhost:9092
       group-id: sample-service-group
       auto-offset-reset: earliest
-      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
-      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
+      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
+      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
     properties:
       schema.registry.url: http://localhost:8081
-
+      specific:
+        avro:
+          reader: true
+      spring:
+        deserializer:
+          key:
+            delegate:
+              class: org.apache.kafka.common.serialization.StringDeserializer
+          value:
+            delegate:
+              class: io.confluent.kafka.serializers.KafkaAvroDeserializer
 dgs:
   reload: true
   graphql:
@@ -37,4 +47,4 @@ dgs:
 
 #app:
 #  topic:
-#    product-topic: sample-topic
+#    product-topic: sample-topic