Fixing Serialization issues

Author: luismr

src/main/java/dev/luismachadoreis/flighttracker/server/ping/infrastructure/config/PingEventConsumerConfig.java

@@ -0,0 +1,38 @@
+package dev.luismachadoreis.flighttracker.server.ping.infrastructure.config;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+
+import java.util.Map;
+
+@Configuration
+public class PingEventConsumerConfig {
+
+    private final Map<String, Object> properties;
+
+    public PingEventConsumerConfig(@Qualifier("kafkaListenerContainerFactoryProperties") Map<String, Object> properties) {
+        this.properties = properties;
+    }
+    
+    @Bean(name = "pingEventConsumerFactory")
+    public ConsumerFactory<String, String> consumerFactory() {
+        return new DefaultKafkaConsumerFactory<>(
+            properties,
+            new StringDeserializer(),
+            new StringDeserializer()
+        );
+    }
+    
+    @Bean(name = "pingEventKafkaListenerContainerFactory")
+    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+        ConcurrentKafkaListenerContainerFactory<String, String> factory =
+            new ConcurrentKafkaListenerContainerFactory<>();
+        factory.setConsumerFactory(consumerFactory());
+        return factory;
+    }
+} 

src/main/java/dev/luismachadoreis/flighttracker/server/ping/infrastructure/pubsub/PingEventConsumer.java

@@ -23,9 +23,13 @@ public class PingEventConsumer {
     /**
      * Consumes ping created events from Kafka and forwards them to WebSocket clients.
      * 
-     * @param message the ping created event
+     * @param message the ping created event as JSON string
      */
-    @KafkaListener(topics = "${spring.kafka.topic.ping-created}", groupId = "${spring.kafka.consumer.group-id}")
+    @KafkaListener(
+        topics = "${spring.kafka.topic.ping-created}", 
+        groupId = "${spring.kafka.consumer.group-id}",
+        containerFactory = "pingEventKafkaListenerContainerFactory"
+    )
     public void consumePingCreated(String message) {
         log.debug("Received ping created event from Kafka: {}", message);
         if (StringUtils.hasText(message)) {
@@ -34,5 +38,4 @@ public void consumePingCreated(String message) {
             log.debug("Skipping empty or null message");
         }
     }
-
 }