feat : use listener instead of calling receiver everytime

Author: rajadilipkolli

kafka-reactor/boot-kafka-reactor-consumer/pom.xml

@@ -15,7 +15,7 @@
     <description>Demo project for Spring Boot</description>
 
     <properties>
-        <java.version>17</java.version>
+        <java.version>21</java.version>
         <springdoc-openapi.version>2.6.0</springdoc-openapi.version>
 
         <spotless.version>2.43.0</spotless.version>
@@ -28,11 +28,12 @@
         </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
         </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-webflux</artifactId>
+            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
         </dependency>
         <dependency>
             <groupId>org.liquibase</groupId>
@@ -42,6 +43,17 @@
             <groupId>org.springframework</groupId>
             <artifactId>spring-jdbc</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>r2dbc-postgresql</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
@@ -51,29 +63,13 @@
             <artifactId>reactor-kafka</artifactId>
             <version>1.3.22</version>
         </dependency>
+
         <dependency>
             <groupId>org.springdoc</groupId>
             <artifactId>springdoc-openapi-starter-webflux-ui</artifactId>
             <version>${springdoc-openapi.version}</version>
         </dependency>
 
-
-        <dependency>
-            <groupId>org.postgresql</groupId>
-            <artifactId>postgresql</artifactId>
-            <scope>runtime</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.postgresql</groupId>
-            <artifactId>r2dbc-postgresql</artifactId>
-            <scope>runtime</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <optional>true</optional>
-        </dependency>
-
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-devtools</artifactId>
@@ -127,14 +123,6 @@
             <plugin>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
-                <configuration>
-                    <excludes>
-                        <exclude>
-                            <groupId>org.projectlombok</groupId>
-                            <artifactId>lombok</artifactId>
-                        </exclude>
-                    </excludes>
-                </configuration>
             </plugin>
             <plugin>
                 <groupId>com.diffplug.spotless</groupId>
@@ -143,7 +131,7 @@
                 <configuration>
                     <java>
                         <palantirJavaFormat>
-                            <version>2.42.0</version>
+                            <version>2.47.0</version>
                         </palantirJavaFormat>
                         <importOrder/>
                         <removeUnusedImports/>

kafka-reactor/boot-kafka-reactor-consumer/src/main/java/com/example/boot/kafka/reactor/config/KafkaConfiguration.java

@@ -1,37 +1,22 @@
 package com.example.boot.kafka.reactor.config;
 
-import com.example.boot.kafka.reactor.entity.MessageDTO;
 import com.example.boot.kafka.reactor.util.AppConstants;
-import java.util.Collections;
-import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.NewTopic;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.kafka.annotation.EnableKafka;
-import reactor.kafka.receiver.KafkaReceiver;
-import reactor.kafka.receiver.ReceiverOptions;
 
 @EnableKafka
 @Configuration(proxyBeanMethods = false)
-@Slf4j
-public class KafkaConfiguration {
+class KafkaConfiguration {
+
+    private static final Logger log = LoggerFactory.getLogger(KafkaConfiguration.class);
 
     @Bean
     NewTopic helloTopic() {
         log.info("Creating helloTopic");
         return new NewTopic(AppConstants.HELLO_TOPIC, 1, (short) 1);
     }
-
-    @Bean
-    KafkaReceiver<Integer, MessageDTO> receiver(KafkaProperties properties) {
-        log.info("Creating receiver");
-        ReceiverOptions<Integer, MessageDTO> receiverOptions = ReceiverOptions.<Integer, MessageDTO>create(
-                        properties.buildConsumerProperties())
-                .subscription(Collections.singleton(AppConstants.HELLO_TOPIC))
-                .addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
-                .addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));
-
-        return KafkaReceiver.create(receiverOptions);
-    }
 }

kafka-reactor/boot-kafka-reactor-consumer/src/main/java/com/example/boot/kafka/reactor/config/SwaggerConfig.java

@@ -7,4 +7,4 @@
 
 @Configuration(proxyBeanMethods = false)
 @OpenAPIDefinition(info = @Info(title = "boot-kafka-reactor-consumer", version = "v1"), servers = @Server(url = "/"))
-public class SwaggerConfig {}
+class SwaggerConfig {}

kafka-reactor/boot-kafka-reactor-consumer/src/main/java/com/example/boot/kafka/reactor/controller/MessageController.java

@@ -2,7 +2,6 @@
 
 import com.example.boot.kafka.reactor.entity.MessageDTO;
 import com.example.boot.kafka.reactor.service.MessageService;
-import lombok.RequiredArgsConstructor;
 import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -11,13 +10,16 @@
 
 @RestController
 @RequestMapping("/messages")
-@RequiredArgsConstructor
 class MessageController {
 
     private final MessageService messageService;
 
+    public MessageController(MessageService messageService) {
+        this.messageService = messageService;
+    }
+
     @GetMapping(produces = MediaType.TEXT_EVENT_STREAM_VALUE)
-    public Flux<MessageDTO> events() {
+    Flux<MessageDTO> events() {
         return messageService.fetchMessages();
     }
 }

kafka-reactor/boot-kafka-reactor-consumer/src/main/java/com/example/boot/kafka/reactor/service/MessageService.java

@@ -2,42 +2,50 @@
 
 import com.example.boot.kafka.reactor.entity.MessageDTO;
 import com.example.boot.kafka.reactor.repository.MessageRepository;
-import java.time.LocalDateTime;
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
+import com.example.boot.kafka.reactor.util.AppConstants;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.kafka.receiver.KafkaReceiver;
-import reactor.kafka.receiver.ReceiverOffset;
 
 @Service
-@RequiredArgsConstructor
-@Slf4j
 public class MessageService {
 
-    private final KafkaReceiver<Integer, MessageDTO> receiver;
+    private static final Logger log = LoggerFactory.getLogger(MessageService.class);
+
     private final MessageRepository messageRepository;
 
-    private Mono<MessageDTO> saveMessage(MessageDTO messageDTO) {
-        return messageRepository.save(messageDTO);
+    public MessageService(MessageRepository messageRepository) {
+        this.messageRepository = messageRepository;
+    }
+
+    @KafkaListener(topics = AppConstants.HELLO_TOPIC, groupId = "reactivekafka")
+    Mono<MessageDTO> listen(
+            @Header(KafkaHeaders.RECEIVED_KEY) Integer key, ConsumerRecord<Integer, MessageDTO> consumerRecord) {
+        ZonedDateTime zdt =
+                ZonedDateTime.ofInstant(Instant.ofEpochMilli(consumerRecord.timestamp()), ZoneId.systemDefault());
+        log.info(
+                "Received message: topic-partition={} offset={} timestamp={} key={} value={}",
+                consumerRecord.partition(),
+                consumerRecord.offset(),
+                zdt,
+                key,
+                consumerRecord.value());
+        return messageRepository.save(consumerRecord.value());
     }
 
     public Flux<MessageDTO> fetchMessages() {
-        return receiver.receive()
-                .map(record -> {
-                    ReceiverOffset offset = record.receiverOffset();
-                    var value = record.value();
-                    log.info(
-                            "Received message: topic-partition={} offset={} timestamp={} key={} value={}",
-                            offset.topicPartition(),
-                            offset.offset(),
-                            LocalDateTime.now(),
-                            record.key(),
-                            value);
-                    offset.acknowledge();
-                    return value;
-                })
-                .flatMap(this::saveMessage);
+        return messageRepository
+                .findAll()
+                .doOnNext(messageDTO -> log.info("Retrieved Message :{}", messageDTO))
+                .doOnError(e -> log.error("Reading Error ", e));
     }
 }

kafka-reactor/boot-kafka-reactor-consumer/src/main/java/com/example/boot/kafka/reactor/util/AppConstants.java

@@ -1,8 +1,9 @@
 package com.example.boot.kafka.reactor.util;
 
-import lombok.experimental.UtilityClass;
-
-@UtilityClass
 public class AppConstants {
     public static final String HELLO_TOPIC = "hello";
+
+    private AppConstants() {
+        throw new UnsupportedOperationException("This is a utility class and cannot be instantiated");
+    }
 }

kafka-reactor/boot-kafka-reactor-consumer/src/test/java/com/example/boot/kafka/reactor/BootKafkaReactorConsumerApplicationTests.java

@@ -5,10 +5,11 @@
 import java.security.SecureRandom;
 import java.time.LocalDateTime;
 import java.util.concurrent.TimeUnit;
-import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.reactive.AutoConfigureWebTestClient;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -23,9 +24,10 @@
 @SpringBootTest(classes = TestBootKafkaReactorConsumerApplication.class)
 @ActiveProfiles("test")
 @AutoConfigureWebTestClient
-@Slf4j
 class BootKafkaReactorConsumerApplicationTests {
 
+    private static final Logger log = LoggerFactory.getLogger(BootKafkaReactorConsumerApplicationTests.class);
+
     @Autowired
     KafkaSender<Integer, MessageDTO> sender;
 

kafka-reactor/boot-kafka-reactor-consumer/src/test/java/com/example/boot/kafka/reactor/TestBootKafkaReactorConsumerApplication.java

@@ -2,7 +2,8 @@
 
 import com.example.boot.kafka.reactor.entity.MessageDTO;
 import java.util.Map;
-import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.test.context.TestConfiguration;
@@ -16,8 +17,9 @@
 import reactor.kafka.sender.SenderOptions;
 
 @TestConfiguration(proxyBeanMethods = false)
-@Slf4j
-public class TestBootKafkaReactorConsumerApplication {
+class TestBootKafkaReactorConsumerApplication {
+
+    private static final Logger log = LoggerFactory.getLogger(TestBootKafkaReactorConsumerApplication.class);
 
     @Bean
     @ServiceConnection