m

pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
-		<version>3.0.3-SNAPSHOT</version>
+		<version>2.7.7</version>
 		<relativePath/> <!-- lookup parent from repository -->
 	</parent>
 	<groupId>com.example</groupId>
@@ -14,18 +14,34 @@
 	<name>kafka</name>
 	<description>Demo project for Spring Webflux and kafka</description>
 	<properties>
-		<java.version>17</java.version>
+		<java.version>1.8</java.version>
 	</properties>
 	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-webflux</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor.kafka</groupId>
+			<artifactId>reactor-kafka</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.kafka</groupId>
 			<artifactId>spring-kafka</artifactId>
 		</dependency>
-
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-clients</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -36,11 +52,6 @@
 			<artifactId>reactor-test</artifactId>
 			<scope>test</scope>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.kafka</groupId>
-			<artifactId>spring-kafka-test</artifactId>
-			<scope>test</scope>
-		</dependency>
 	</dependencies>
 
 	<build>
@@ -51,41 +62,5 @@
 			</plugin>
 		</plugins>
 	</build>
-	<repositories>
-		<repository>
-			<id>spring-milestones</id>
-			<name>Spring Milestones</name>
-			<url>https://repo.spring.io/milestone</url>
-			<snapshots>
-				<enabled>false</enabled>
-			</snapshots>
-		</repository>
-		<repository>
-			<id>spring-snapshots</id>
-			<name>Spring Snapshots</name>
-			<url>https://repo.spring.io/snapshot</url>
-			<releases>
-				<enabled>false</enabled>
-			</releases>
-		</repository>
-	</repositories>
-	<pluginRepositories>
-		<pluginRepository>
-			<id>spring-milestones</id>
-			<name>Spring Milestones</name>
-			<url>https://repo.spring.io/milestone</url>
-			<snapshots>
-				<enabled>false</enabled>
-			</snapshots>
-		</pluginRepository>
-		<pluginRepository>
-			<id>spring-snapshots</id>
-			<name>Spring Snapshots</name>
-			<url>https://repo.spring.io/snapshot</url>
-			<releases>
-				<enabled>false</enabled>
-			</releases>
-		</pluginRepository>
-	</pluginRepositories>
 
 </project>

src/main/java/com/example/kafka/config/KafkaProducerConfig.java

+package com.example.kafka.config;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Component
+@Data
+public class KafkaProducerConfig {
+
+    @Value("${bootstrap.server}")
+    private String bootstrapper;
+
+}

src/main/java/com/example/kafka/controller/OrderController.java

+package com.example.kafka.controller;
+
+import com.example.kafka.entity.Order;
+import com.example.kafka.serviceImpl.OrderServiceImpl;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequestMapping("/v1/order")
+public class OrderController {
+
+    @Autowired
+    private OrderServiceImpl orderService;
+
+    @PostMapping("/save")
+    @ResponseStatus(HttpStatus.ACCEPTED)
+    public Mono<Order> saveOrderDetails(@RequestBody Order order){
+        return orderService.saveOrderIntoDB(order);
+    }
+
+}

src/main/java/com/example/kafka/entity/Order.java

+package com.example.kafka.entity;
+
+import lombok.*;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+@Data
+@ToString
+@EqualsAndHashCode(of={"id","name","amount"})
+@AllArgsConstructor
+@NoArgsConstructor
+@Document(value = "orders")
+public class Order {
+    private String id;
+    private String name;
+    private double amount;
+}

src/main/java/com/example/kafka/repository/OrderRepository.java

+package com.example.kafka.repository;
+
+import com.example.kafka.entity.Order;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface OrderRepository extends ReactiveMongoRepository<Order,String> {
+}

src/main/java/com/example/kafka/service/OrderService.java

+package com.example.kafka.service;
+
+import com.example.kafka.entity.Order;
+import reactor.core.publisher.Mono;
+
+public interface OrderService {
+    Mono<Order> saveOrderIntoDB(Order order);
+}

src/main/java/com/example/kafka/serviceImpl/OrderServiceImpl.java

+package com.example.kafka.serviceImpl;
+
+import com.example.kafka.entity.Order;
+import com.example.kafka.repository.OrderRepository;
+import com.example.kafka.service.OrderService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+@Service
+public class OrderServiceImpl implements OrderService {
+
+    @Autowired
+    private OrderRepository orderRepository;
+    @Override
+    public Mono<Order> saveOrderIntoDB(Order order) {
+        return orderRepository.save(order);
+    }
+}

src/main/resources/application.properties

+bootstrap.server=localhost:9092
+kafka.clientId=rewards-generator
 
+spring.webflux.base-path=/api
+spring.application.name=spring-webflux-kafka
+
+data.mongodb.authentication-database=admin
+data.mongodb.uri=mongodb://localhost:27017/test
+data.mongodb.database=test
+
+server.port:9000
+
+logging.level.io.reflectoring: DEBUG
+logging.level.org.springframework.web: INFO
+logging.level.org.springframework.data.mongodb.core.ReactiveMongoTemplate=DEBUG
+logging.level.reactor.netty.http.client=DEBUG
\ No newline at end of file

src/test/java/com/example/kafka/KafkaApplicationTests.java

-package com.example.kafka;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-
-@SpringBootTest
-class KafkaApplicationTests {
-
-	@Test
-	void contextLoads() {
-	}
-
-}