feat : use customizer to configure deadletter config

Author: rajadilipkolli

order-service/src/main/java/com/example/orderservice/config/kafka/KafkaStreamsConfig.java

@@ -13,8 +13,7 @@ Licensed under MIT License Copyright (c) 2023 Raja Kolli.
 import com.example.common.dtos.OrderDto;
 import com.example.orderservice.services.OrderManageService;
 import java.time.Duration;
-import java.util.Map;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
+import java.util.Properties;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
@@ -31,19 +30,16 @@ Licensed under MIT License Copyright (c) 2023 Raja Kolli.
 import org.apache.kafka.streams.state.Stores;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafkaStreams;
-import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
-import org.springframework.kafka.config.KafkaStreamsConfiguration;
 import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
 import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler;
 import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.util.Assert;
 
 @Configuration(proxyBeanMethods = false)
 @EnableKafkaStreams
@@ -58,32 +54,23 @@ class KafkaStreamsConfig {
     }
 
     @Bean
-    StreamsBuilderFactoryBeanConfigurer configurer() {
+    StreamsBuilderFactoryBeanConfigurer configurer(
+            DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
         return factoryBean -> {
             factoryBean.setStateListener(
                     (newState, oldState) ->
                             log.info("State transition from {} to {} ", oldState, newState));
+            Properties streamsConfiguration = factoryBean.getStreamsConfiguration();
+            Assert.notNull(streamsConfiguration, "streamsConfiguration must not be null");
+            streamsConfiguration.put(
+                    StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+                    RecoveringDeserializationExceptionHandler.class);
+            streamsConfiguration.put(
+                    RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER,
+                    deadLetterPublishingRecoverer);
         };
     }
 
-    @Bean(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
-    KafkaStreamsConfiguration defaultKafkaStreamsConfig(
-            KafkaConnectionDetails connectionDetails,
-            KafkaProperties kafkaProperties,
-            DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
-        Map<String, Object> properties = kafkaProperties.buildStreamsProperties(null);
-        properties.put(
-                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-                connectionDetails.getStreamsBootstrapServers());
-        properties.put(
-                StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
-                RecoveringDeserializationExceptionHandler.class);
-        properties.put(
-                RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER,
-                deadLetterPublishingRecoverer);
-        return new KafkaStreamsConfiguration(properties);
-    }
-
     @Bean
     DeadLetterPublishingRecoverer deadLetterPublishingRecoverer(
             ProducerFactory<byte[], byte[]> producerFactory) {

order-service/src/test/java/com/example/orderservice/TestOrderServiceApplication.java

@@ -10,14 +10,12 @@ Licensed under MIT License Copyright (c) 2023-2024 Raja Kolli.
 import com.example.orderservice.common.PostGreSQLContainer;
 import com.example.orderservice.utils.AppConstants;
 import org.springframework.boot.SpringApplication;
-import org.springframework.boot.testcontainers.context.ImportTestcontainers;
 
-@ImportTestcontainers(PostGreSQLContainer.class)
 public class TestOrderServiceApplication {
 
     public static void main(String[] args) {
         SpringApplication.from(OrderServiceApplication::main)
-                .with(ContainersConfig.class)
+                .with(ContainersConfig.class, PostGreSQLContainer.class)
                 .withAdditionalProfiles(AppConstants.PROFILE_LOCAL)
                 .run(args);
     }

order-service/src/test/java/com/example/orderservice/common/AbstractIntegrationTest.java

@@ -15,7 +15,6 @@ Licensed under MIT License Copyright (c) 2021-2023 Raja Kolli.
 import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.TestConfiguration;
-import org.springframework.boot.testcontainers.context.ImportTestcontainers;
 import org.springframework.context.annotation.Bean;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.web.servlet.MockMvc;
@@ -24,8 +23,7 @@ Licensed under MIT License Copyright (c) 2021-2023 Raja Kolli.
 @SpringBootTest(
         webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
         properties = {"spring.cloud.config.enabled=false"},
-        classes = ContainersConfig.class)
-@ImportTestcontainers(PostGreSQLContainer.class)
+        classes = {ContainersConfig.class, PostGreSQLContainer.class})
 @AutoConfigureMockMvc
 @AutoConfigureObservability
 public abstract class AbstractIntegrationTest extends ContainerInitializer {

order-service/src/test/java/com/example/orderservice/common/PostGreSQLContainer.java

@@ -6,16 +6,19 @@ Licensed under MIT License Copyright (c) 2023 Raja Kolli.
 
 package com.example.orderservice.common;
 
-import java.util.Collections;
+import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Bean;
 import org.testcontainers.containers.PostgreSQLContainer;
 import org.testcontainers.utility.DockerImageName;
 
-public interface PostGreSQLContainer {
+@TestConfiguration(proxyBeanMethods = false)
+public class PostGreSQLContainer {
 
     @ServiceConnection
-    PostgreSQLContainer<?> postgreSQLContainer =
-            new PostgreSQLContainer<>(DockerImageName.parse("postgres").withTag("17-alpine"))
-                    .withReuse(true)
-                    .withTmpFs(Collections.singletonMap("/var/lib/postgresql/data", "rw"));
+    @Bean
+    PostgreSQLContainer<?> postgreSQLContainer() {
+        return new PostgreSQLContainer<>(DockerImageName.parse("postgres").withTag("17-alpine"))
+                .withReuse(true);
+    }
 }

order-service/src/test/java/com/example/orderservice/repositories/OrderRepositoryTest.java

@@ -17,19 +17,17 @@ Licensed under MIT License Copyright (c) 2023-2024 Raja Kolli.
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
 import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;
-import org.springframework.boot.testcontainers.context.ImportTestcontainers;
+import org.springframework.context.annotation.Import;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.test.context.ActiveProfiles;
 
 @ActiveProfiles({PROFILE_TEST})
-@ImportTestcontainers(PostGreSQLContainer.class)
+@Import(PostGreSQLContainer.class)
 @DataJpaTest(properties = {"spring.jpa.hibernate.ddl-auto=validate"})
-@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
 class OrderRepositoryTest {
 
     @Autowired private OrderRepository orderRepository;