feat(spring-kafka-example): add Producer kafka test and custom exception handling

Author: igorcampos-dev

spring-kafka-example/src/main/java/com/io/example/exception/DeserializationException.java

@@ -0,0 +1,9 @@
+package com.io.example.exception;
+
+public class DeserializationException extends RuntimeException {
+
+    public DeserializationException(String message) {
+        super(message);
+    }
+
+}

spring-kafka-example/src/main/java/com/io/example/exception/SerializationException.java

@@ -0,0 +1,9 @@
+package com.io.example.exception;
+
+public class SerializationException extends RuntimeException {
+
+  public SerializationException(String message) {
+        super(message);
+    }
+
+}

spring-kafka-example/src/main/java/com/io/example/mapper/JsonMapper.java

@@ -2,6 +2,8 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.io.example.exception.SerializationException;
+import com.io.example.exception.DeserializationException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -18,7 +20,7 @@ public String toJsonString(Object obj) {
             return objectMapper.writeValueAsString(obj);
         } catch (JsonProcessingException e) {
             log.error("error converting an object of type {} to json. cause of the error: {}", obj.getClass(), e.getMessage());
-            throw new RuntimeException("JSON serialization failed", e);
+            throw new SerializationException("JSON serialization failed, cause: " + e.getMessage());
         }
     }
 
@@ -27,7 +29,7 @@ public <T> T toObject(String jsonBody, Class<T> typeClass){
             return objectMapper.readValue(jsonBody, typeClass);
         } catch (JsonProcessingException e) {
             log.error("error converting an String to object. cause of the error: {}", e.getMessage());
-            throw new IllegalArgumentException("Object deserialization failed for target type: " + typeClass.getSimpleName(), e);
+            throw new DeserializationException("Object deserialization failed for target type: " + typeClass.getSimpleName() + ", cause: " + e.getMessage());
         }
     }
 

spring-kafka-example/src/test/java/com/io/example/KafkaProducerServiceImplTest.java

@@ -0,0 +1,84 @@
+package com.io.example;
+
+import com.io.example.controller.dto.request.MessageRequestDtoRequest;
+import com.io.example.mapper.JsonMapper;
+import com.io.example.producer.KafkaProducerServiceImpl;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@ActiveProfiles("test")
+@EmbeddedKafka(partitions = 1, topics = {"spring-kafka-example-simple-topic-test"})
+public class KafkaProducerServiceImplTest {
+
+    @SuppressWarnings("unused")
+    @Value("${spring.kafka.topics.topic-1}")
+    private String topic_1;
+
+    @SuppressWarnings("unused")
+    @Value("${kafka.group-id}")
+    private String group_id;
+
+    @Autowired
+    @SuppressWarnings("unused")
+    private KafkaProducerServiceImpl producer;
+
+    @Autowired
+    @SuppressWarnings("unused")
+    private JsonMapper mapper;
+
+    @Autowired
+    @SuppressWarnings("unused")
+    private EmbeddedKafkaBroker embeddedKafkaBroker;
+
+    private MessageRequestDtoRequest message;
+
+    @BeforeEach
+    void init(){
+        message = new MessageRequestDtoRequest("Teste Kafka");
+    }
+
+    @Test
+    void testSendMessage() {
+
+        producer.sendMessage(message);
+
+        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(this.getProps())) {
+            consumer.subscribe(Collections.singleton(topic_1));
+            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
+
+            assertThat(records.count()).isGreaterThan(0);
+
+            ConsumerRecord<String, String> record = records.iterator().next();
+            MessageRequestDtoRequest receivedMessage = mapper.toObject(record.value(), MessageRequestDtoRequest.class);
+            assertThat(receivedMessage.message()).isEqualTo("Teste Kafka");
+
+        }
+    }
+
+    private Properties getProps(){
+        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(group_id, "true", embeddedKafkaBroker);
+        Properties props = new Properties();
+        props.putAll(consumerProps);
+        return props;
+    }
+
+
+}

spring-kafka-example/src/test/resources/application-test.yml

@@ -0,0 +1,24 @@
+spring:
+
+  application:
+    name: spring-kafka-example-test
+
+  kafka:
+    topics:
+      topic-1: spring-kafka-example-simple-topic-test
+    bootstrap-servers: localhost:9093
+    consumer:
+      group-id: spring-kafka-example-group-test
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    listener:
+      missing-topics-fatal: false
+
+logging:
+
+  pattern:
+    console: "%d{yyyy-MM-dd'T'HH:mm:ss} | ${spring.application.name} | %class{30} | %level | %m%n"
+
+kafka:
+  group-id: group-test