chore(spring-kafka-test): general improvements in the project

Author: igorcampos-dev

.github/workflows/spring-kafka-example.yml

@@ -68,11 +68,19 @@ jobs:
           echo "Checking the health of the containers..."
 
           SERVICES="${{ steps.services.outputs.services }}"
+
           for service in $SERVICES; do
-            status=$(docker inspect -f '{{.State.Running}}' "$service" || echo "false")
-            echo "$service status: $status"
-            if [ "$status" != "true" ]; then
-              echo "::error ::Service $service is not running."
-              exit 1
+            running=$(docker inspect -f '{{.State.Running}}' "$service" 2>/dev/null || echo "false")
+
+            if [ "$running" = "true" ]; then
+              echo "$service is running ✔️"
+            else
+              exit_code=$(docker inspect -f '{{.State.ExitCode}}' "$service" 2>/dev/null || echo "1")
+              if [ "$exit_code" = "0" ]; then
+                echo "$service is stopped but exited with code 0 ✔️"
+              else
+                echo "::error ::$service is stopped and exited with code $exit_code ❌"
+                exit 1
+              fi
             fi
-          done
+          done

spring-kafka-example/compose.yaml

@@ -9,25 +9,59 @@ services:
     environment:
       SERVER_PORT: "80"
       SPRING_PROFILES_ACTIVE: "default"
-      KAFKA_SERVER_PORT: kafka:9092
+      KAFKA_TOPIC_1: spring-kafka-example-simple-topic
+      KAFKA_BOOTSTRAP_SERVERS: kafka:9092
+      KAFKA_GROUP_ID: spring-kafka-example-group
+      KAFKA_AUTO_OFFSET_RESET: earliest
+      KAFKA_KEY_DESERIALIZER: org.apache.kafka.common.serialization.StringDeserializer
+      KAFKA_VALUE_DESERIALIZER: org.apache.kafka.common.serialization.StringDeserializer
+      KAFKA_MISSING_TOPICS_FATAL: false
     depends_on:
       - kafka
+    networks:
+      - kafka-net
 
   kafka:
     image: apache/kafka:4.0.0
     container_name: kafka
     ports:
       - "9092:9092"
+      - "9093:9093"
     environment:
       KAFKA_NODE_ID: 1
       KAFKA_PROCESS_ROLES: broker,controller
-      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
-      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9093
+      KAFKA_LISTENERS: CONTROLLER://0.0.0.0:9094,PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:9093
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
       KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9094
       KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
-      CLUSTER_ID: qGyb4Z0XQpeoKgUXYfCCLw
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
+    networks:
+      - kafka-net
+
+  kafka-init:
+    container_name: kafka-init
+    image: apache/kafka:4.0.0
+    depends_on:
+      kafka:
+        condition: service_started
+    command: [ "/bin/bash", "-c", "/create_topic.sh" ]
+    environment:
+      KAFKA_HOST: kafka
+      KAFKA_PORT: 9092
+      KAFKA_TOPIC_CREATE: spring-kafka-example-simple-topic
+    volumes:
+      - type: bind
+        source: ./docker/create_topic.sh
+        target: /create_topic.sh
+    init: true
+    networks:
+      - kafka-net
+
+networks:
+  kafka-net:
+    driver: bridge
+

spring-kafka-example/docker/create_topic.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+echo "Waiting for Kafka on port $KAFKA_PORT..."
+while ! nc -z $KAFKA_HOST $KAFKA_PORT; do
+  sleep 2
+done
+
+/opt/kafka/bin/kafka-topics.sh \
+  --create \
+  --topic "$KAFKA_TOPIC_CREATE" \
+  --bootstrap-server $KAFKA_HOST:$KAFKA_PORT \
+  --partitions 1 \
+  --replication-factor 1
+
+echo "Topic $KAFKA_TOPIC_CREATE created successfully!"

spring-kafka-example/src/main/java/com/io/example/config/JacksonConfig.java

@@ -0,0 +1,20 @@
+package com.io.example.config;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class JacksonConfig {
+
+    @Bean
+    public ObjectMapper objectMapper() {
+        ObjectMapper mapper = new ObjectMapper();
+        mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+        mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+        mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        return mapper;
+    }
+}

spring-kafka-example/src/main/java/com/io/example/consumer/KafkaConsumerService.java

@@ -2,5 +2,5 @@
 
 @SuppressWarnings("unused")
 public interface KafkaConsumerService {
-    void consume(String mensagem);
+    void consume(String message);
 }

spring-kafka-example/src/main/java/com/io/example/consumer/KafkaConsumerServiceImpl.java

@@ -1,17 +1,24 @@
 package com.io.example.consumer;
 
+import com.io.example.controller.dto.request.MessageRequestDtoRequest;
+import com.io.example.mapper.JsonMapper;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Service;
 
 @Slf4j
 @Service
+@RequiredArgsConstructor
 @SuppressWarnings("unused")
 public class KafkaConsumerServiceImpl implements KafkaConsumerService {
 
+    private final JsonMapper mapper;
+
     @KafkaListener(topics = "${spring.kafka.topics.topic-1}" , groupId = "${spring.kafka.consumer.group-id}")
     public void consume(String message) {
-        log.info("Message receive: {} ", message);
+        MessageRequestDtoRequest messageRequestDtoRequest = mapper.toObject(message, MessageRequestDtoRequest.class);
+        log.info("Message receive: {} ", messageRequestDtoRequest.toString());
     }
 
 }

spring-kafka-example/src/main/java/com/io/example/controller/ExampleController.java

@@ -1,7 +1,7 @@
 package com.io.example.controller;
 
 import com.io.example.controller.dto.request.MessageRequestDtoRequest;
-import com.io.example.controller.dto.response.GetDtoResponse;
+import com.io.example.controller.dto.response.SimpleTopicDtoResponse;
 import com.io.example.producer.KafkaProducerService;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.HttpStatus;
@@ -15,10 +15,10 @@ public class ExampleController {
 
     private final KafkaProducerService kafkaProducerService;
 
-    @PostMapping("/send")
-    public ResponseEntity<GetDtoResponse> sendMessage(@RequestBody MessageRequestDtoRequest dto){
-        this.kafkaProducerService.sendMessage(dto.message());
-        return ResponseEntity.status(HttpStatus.OK).body(new GetDtoResponse("Message sent successfully!"));
+    @PostMapping("/send/simple-topic")
+    public ResponseEntity<SimpleTopicDtoResponse> sendMessageToSimpleTopic(@RequestBody MessageRequestDtoRequest dto){
+        this.kafkaProducerService.sendMessage(dto);
+        return ResponseEntity.status(HttpStatus.OK).body(new SimpleTopicDtoResponse("Message sent successfully!"));
     }
 
 }

spring-kafka-example/src/main/java/com/io/example/controller/dto/response/GetDtoResponse.java

@@ -0,0 +1,3 @@
+package com.io.example.controller.dto.response;
+
+public record SimpleTopicDtoResponse(String message) {}

spring-kafka-example/src/main/java/com/io/example/controller/dto/response/SimpleTopicDtoResponse.java

@@ -1,6 +1,8 @@
 package com.io.example.producer;
 
+import com.io.example.controller.dto.request.MessageRequestDtoRequest;
+
 @FunctionalInterface
 public interface KafkaProducerService {
-    void sendMessage(String mensagem);
+    void sendMessage(MessageRequestDtoRequest mensagem);
 }