codebase/synchronous-communication-with-apache-kafka-using-replyingkafkatemplate [BAEL-6137] (#18405)

* add codebase

* add test case

* add missing dependency

* minor updates

* add test scope to test dependency

* remove test application

* remove reply topic name in @sendto

* remove @EnableKafka

* upgrade kafka docker version

* add validation on kafka configuration properties

* specify package explicitly in trusted packages

* upgrade spring-boot version

* add controller for local testing

Author: hardikSinghBehl

spring-kafka-4/pom.xml

@@ -17,24 +17,26 @@
     <dependencies>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter</artifactId>
+            <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-web</artifactId>
+            <artifactId>spring-boot-starter-validation</artifactId>
         </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
-            <version>${spring-kafka.version}</version>
         </dependency>
+
+        <!-- test dependencies -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-webflux</artifactId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
+            <artifactId>spring-boot-testcontainers</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -52,10 +54,6 @@
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
     </dependencies>
 
     <build>
@@ -72,8 +70,7 @@
 
     <properties>
         <java.version>21</java.version>
-        <spring-kafka.version>3.1.2</spring-kafka.version>
-        <spring-boot.version>3.2.2</spring-boot.version>
+        <spring-boot.version>3.4.4</spring-boot.version>
     </properties>
 
 </project>

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/Application.java

@@ -0,0 +1,15 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.PropertySource;
+
+@SpringBootApplication
+@PropertySource("classpath:application-synchronous-kafka.properties")
+class Application {
+
+    public static void main(String[] args) {
+        SpringApplication.run(Application.class, args);
+    }
+
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/KafkaConfiguration.java

@@ -0,0 +1,64 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.KafkaMessageListenerContainer;
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
+
+import java.time.Duration;
+
+@Configuration
+@EnableConfigurationProperties(SynchronousKafkaProperties.class)
+class KafkaConfiguration {
+
+    private final SynchronousKafkaProperties synchronousKafkaProperties;
+
+    KafkaConfiguration(SynchronousKafkaProperties synchronousKafkaProperties) {
+        this.synchronousKafkaProperties = synchronousKafkaProperties;
+    }
+
+    @Bean
+    KafkaMessageListenerContainer<String, NotificationDispatchResponse> kafkaMessageListenerContainer(
+        ConsumerFactory<String, NotificationDispatchResponse> consumerFactory
+    ) {
+        String replyTopic = synchronousKafkaProperties.replyTopic();
+        ContainerProperties containerProperties = new ContainerProperties(replyTopic);
+        return new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
+    }
+
+    @Bean
+    ReplyingKafkaTemplate<String, NotificationDispatchRequest, NotificationDispatchResponse> replyingKafkaTemplate(
+        ProducerFactory<String, NotificationDispatchRequest> producerFactory,
+        KafkaMessageListenerContainer<String, NotificationDispatchResponse> kafkaMessageListenerContainer
+    ) {
+        Duration replyTimeout = synchronousKafkaProperties.replyTimeout();
+        var replyingKafkaTemplate = new ReplyingKafkaTemplate<>(producerFactory, kafkaMessageListenerContainer);
+        replyingKafkaTemplate.setDefaultReplyTimeout(replyTimeout);
+        return replyingKafkaTemplate;
+    }
+
+    @Bean
+    KafkaTemplate<String, NotificationDispatchResponse> kafkaTemplate(ProducerFactory<String, NotificationDispatchResponse> producerFactory) {
+        return new KafkaTemplate<>(producerFactory);
+    }
+
+    @Bean
+    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, NotificationDispatchRequest>> kafkaListenerContainerFactory(
+        ConsumerFactory<String, NotificationDispatchRequest> consumerFactory,
+        KafkaTemplate<String, NotificationDispatchResponse> kafkaTemplate
+    ) {
+        var factory = new ConcurrentKafkaListenerContainerFactory<String, NotificationDispatchRequest>();
+        factory.setConsumerFactory(consumerFactory);
+        factory.setReplyTemplate(kafkaTemplate);
+        return factory;
+    }
+
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/NotificationDispatchController.java

@@ -0,0 +1,29 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.concurrent.ExecutionException;
+
+@RestController
+@RequestMapping("/api/v1")
+class NotificationDispatchController {
+
+    private final NotificationDispatchService notificationDispatchService;
+
+    NotificationDispatchController(NotificationDispatchService notificationDispatchService) {
+        this.notificationDispatchService = notificationDispatchService;
+    }
+
+    @PostMapping(value = "/notification")
+    ResponseEntity<NotificationDispatchResponse> dispatch(
+        @RequestBody NotificationDispatchRequest notificationDispatchRequest
+    ) throws ExecutionException, InterruptedException {
+        NotificationDispatchResponse response = notificationDispatchService.dispatch(notificationDispatchRequest);
+        return ResponseEntity.ok(response);
+    }
+
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/NotificationDispatchListener.java

@@ -0,0 +1,20 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.stereotype.Component;
+
+import java.util.UUID;
+
+@Component
+class NotificationDispatchListener {
+
+    @SendTo
+    @KafkaListener(topics = "${com.baeldung.kafka.synchronous.request-topic}")
+    NotificationDispatchResponse listen(NotificationDispatchRequest notificationDispatchRequest) {
+        // ... processing logic
+        UUID notificationId = UUID.randomUUID();
+        return new NotificationDispatchResponse(notificationId);
+    }
+
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/NotificationDispatchRequest.java

@@ -0,0 +1,4 @@
+package com.baeldung.kafka.synchronous;
+
+record NotificationDispatchRequest(String emailId, String content) {
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/NotificationDispatchResponse.java

@@ -0,0 +1,6 @@
+package com.baeldung.kafka.synchronous;
+
+import java.util.UUID;
+
+public record NotificationDispatchResponse(UUID notificationId) {
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/NotificationDispatchService.java

@@ -0,0 +1,30 @@
+package com.baeldung.kafka.synchronous;
+
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.ExecutionException;
+
+@Service
+@EnableConfigurationProperties(SynchronousKafkaProperties.class)
+class NotificationDispatchService {
+
+    private final SynchronousKafkaProperties synchronousKafkaProperties;
+    private final ReplyingKafkaTemplate<String, NotificationDispatchRequest, NotificationDispatchResponse> replyingKafkaTemplate;
+
+    NotificationDispatchService(SynchronousKafkaProperties synchronousKafkaProperties, ReplyingKafkaTemplate<String, NotificationDispatchRequest, NotificationDispatchResponse> replyingKafkaTemplate) {
+        this.synchronousKafkaProperties = synchronousKafkaProperties;
+        this.replyingKafkaTemplate = replyingKafkaTemplate;
+    }
+
+    NotificationDispatchResponse dispatch(NotificationDispatchRequest notificationDispatchRequest) throws ExecutionException, InterruptedException {
+        String requestTopic = synchronousKafkaProperties.requestTopic();
+        ProducerRecord<String, NotificationDispatchRequest> producerRecord = new ProducerRecord<>(requestTopic, notificationDispatchRequest);
+
+        var requestReplyFuture = replyingKafkaTemplate.sendAndReceive(producerRecord);
+        return requestReplyFuture.get().value();
+    }
+
+}

spring-kafka-4/src/main/java/com/baeldung/kafka/synchronous/SynchronousKafkaProperties.java

@@ -0,0 +1,24 @@
+package com.baeldung.kafka.synchronous;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import org.hibernate.validator.constraints.time.DurationMax;
+import org.hibernate.validator.constraints.time.DurationMin;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+import java.time.Duration;
+
+@Validated
+@ConfigurationProperties(prefix = "com.baeldung.kafka.synchronous")
+record SynchronousKafkaProperties(
+    @NotBlank
+    String requestTopic,
+
+    @NotBlank
+    String replyTopic,
+
+    @NotNull @DurationMin(seconds = 10) @DurationMax(minutes = 2)
+    Duration replyTimeout
+) {
+}

spring-kafka-4/src/main/resources/application-synchronous-kafka.properties

@@ -0,0 +1,15 @@
+spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS}
+
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.group-id=synchronous-kafka-group
+spring.kafka.consumer.properties.spring.json.trusted.packages=com.baeldung.kafka.synchronous
+
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
+
+spring.kafka.properties.allow.auto.create.topics=true
+
+com.baeldung.kafka.synchronous.request-topic=notification-dispatch-request
+com.baeldung.kafka.synchronous.reply-topic=notification-dispatch-response
+com.baeldung.kafka.synchronous.reply-timeout=30s

spring-kafka-4/src/test/java/com/baeldung/kafka/synchronous/SynchronousKafkaLiveTest.java

@@ -0,0 +1,29 @@
+package com.baeldung.kafka.synchronous;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.context.annotation.Import;
+
+import java.util.concurrent.ExecutionException;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+@SpringBootTest
+@Import(TestcontainersConfiguration.class)
+class SynchronousKafkaLiveTest {
+
+    @Autowired
+    private NotificationDispatchService notificationDispatchService;
+
+    @Test
+    void whenNotificationRequestSent_thenReplyReceived() throws ExecutionException, InterruptedException {
+        NotificationDispatchRequest request = new NotificationDispatchRequest("[email protected]", "test-content");
+
+        NotificationDispatchResponse response = notificationDispatchService.dispatch(request);
+
+        assertThat(response.notificationId())
+            .isNotNull();
+    }
+
+}

spring-kafka-4/src/test/java/com/baeldung/kafka/synchronous/TestApplication.java

@@ -0,0 +1,13 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.boot.SpringApplication;
+
+class TestApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.from(Application::main)
+            .with(TestcontainersConfiguration.class)
+            .run(args);
+    }
+
+}

spring-kafka-4/src/test/java/com/baeldung/kafka/synchronous/TestcontainersConfiguration.java

@@ -0,0 +1,17 @@
+package com.baeldung.kafka.synchronous;
+
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Bean;
+import org.testcontainers.kafka.KafkaContainer;
+
+@TestConfiguration(proxyBeanMethods = false)
+class TestcontainersConfiguration {
+
+    @Bean
+    @ServiceConnection
+    KafkaContainer kafkaContainer() {
+        return new KafkaContainer("apache/kafka:4.0.0");
+    }
+
+}