chore(spring-kafka-test): aditionate producer,consumer, compose configs, global handler exception,logs and lombok in the project

Author: igorcampos-dev

spring-kafka-example/compose.yaml

@@ -9,10 +9,13 @@ services:
     environment:
       SERVER_PORT: "80"
       SPRING_PROFILES_ACTIVE: "default"
+      KAFKA_SERVER_PORT: kafka:9092
+    depends_on:
+      - kafka
 
   kafka:
+    image: apache/kafka:4.0.0
     container_name: kafka
-    image: confluentinc/cp-kafka:8.0.0
     ports:
       - "9092:9092"
     environment:
@@ -26,3 +29,5 @@ services:
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
       KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
       CLUSTER_ID: qGyb4Z0XQpeoKgUXYfCCLw
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'

spring-kafka-example/pom.xml

@@ -22,6 +22,7 @@
 		<java.version>21</java.version>
 		<spring.boot.version>3.5.4</spring.boot.version>
 		<spring.kafka.version>3.3.8</spring.kafka.version>
+		<lombok.version>1.18.38</lombok.version>
 	</properties>
 
 	<dependencies>
@@ -38,6 +39,13 @@
 			<version>${spring.kafka.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+			<version>${lombok.version}</version>
+		</dependency>
+
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -55,15 +63,38 @@
 	</dependencies>
 
 	<build>
+
 		<plugins>
 
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<annotationProcessorPaths>
+						<path>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</path>
+					</annotationProcessorPaths>
+				</configuration>
+			</plugin>
+
 			<plugin>
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
 				<version>${spring.boot.version}</version>
+				<configuration>
+					<excludes>
+						<exclude>
+							<groupId>org.projectlombok</groupId>
+							<artifactId>lombok</artifactId>
+						</exclude>
+					</excludes>
+				</configuration>
 			</plugin>
 
 		</plugins>
+
 	</build>
 
 </project>

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

@@ -0,0 +1,6 @@
+package com.io.example.consumer;
+
+@SuppressWarnings("unused")
+public interface KafkaConsumerService {
+    void consume(String mensagem);
+}

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

@@ -0,0 +1,17 @@
+package com.io.example.consumer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@SuppressWarnings("unused")
+public class KafkaConsumerServiceImpl implements KafkaConsumerService {
+
+    @KafkaListener(topics = "${spring.kafka.topics.topic-1}" , groupId = "${spring.kafka.consumer.group-id}")
+    public void consume(String message) {
+        log.info("Message receive: {} ", message);
+    }
+
+}

spring-kafka-example/src/main/java/com/io/example/controller/dto/request/MessageRequestDtoRequest.java

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

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 GetDtoResponse(String message) {}

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

@@ -0,0 +1,24 @@
+package com.io.example.exception.handler;
+
+import com.io.example.exception.responseBody.Error;
+import jakarta.servlet.http.HttpServletRequest;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+
+@Slf4j
+@ControllerAdvice
+@SuppressWarnings("unused")
+public class GlobalExceptionHandler {
+
+    @ExceptionHandler(RuntimeException.class)
+    public ResponseEntity<ResponseEntity<Error>> handleRuntimeException(RuntimeException ex, HttpServletRequest s) {
+        log.error("HandleRuntimeException= {}", ex.getMessage());
+        return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
+                             .body(Error.response(ex.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR, s));
+    }
+
+
+}

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

@@ -0,0 +1,45 @@
+package com.io.example.exception.responseBody;
+
+import jakarta.servlet.ServletRequest;
+import jakarta.servlet.http.HttpServletRequest;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.server.ServerWebExchange;
+
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class Error {
+    private String timestamp;
+    private String path;
+    private Integer status;
+    private String error;
+
+    public static ResponseEntity<Error> response(String message, HttpStatus status, HttpServletRequest uri){
+        return ResponseEntity
+                .status(status)
+                .body(Error.builder()
+                           .timestamp(getInstantNow())
+                           .path(uri.getRequestURI())
+                           .status(status.value())
+                           .error(message)
+                           .build());
+    }
+
+    private static String getInstantNow() {
+        return Instant.now()
+                      .atZone(ZoneId.systemDefault())
+                      .format(DateTimeFormatter.ofPattern("dd-MM-yyyy HH:mm:ss"));
+
+    }
+
+}

spring-kafka-example/src/main/java/com/io/example/producer/KafkaProducerService.java

@@ -0,0 +1,6 @@
+package com.io.example.producer;
+
+@FunctionalInterface
+public interface KafkaProducerService {
+    void sendMessage(String mensagem);
+}

spring-kafka-example/src/main/java/com/io/example/producer/KafkaProducerServiceImpl.java

@@ -0,0 +1,28 @@
+package com.io.example.producer;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@SuppressWarnings("unused")
+public class KafkaProducerServiceImpl implements KafkaProducerService {
+
+    @SuppressWarnings("unused")
+    @Value("${spring.kafka.topics.topic-1}")
+    private String topic;
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    @Override
+    public void sendMessage(String message) {
+        log.info("Preparing to send message to Kafka topic: {}", topic);
+        kafkaTemplate.send(topic, message);
+        log.info("Message successfully dispatched: {}", message);
+    }
+
+}