integrating kafka

src/main/java/com/nisum/webfluxmongodbordermanagement/config/KafkaConsumerConfig.java

@@ -22,6 +22,7 @@ public class KafkaConsumerConfig {
 
     @Bean
     public ConsumerFactory<String, String> consumerFatory() {
+
         Map<String, Object> config = new HashMap<>();
         config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

src/main/java/com/nisum/webfluxmongodbordermanagement/config/KafkaListeners.java

@@ -9,6 +9,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 public class KafkaListeners {
+
     @Value("${order_topic}")
     private String topicName;
 

src/main/java/com/nisum/webfluxmongodbordermanagement/config/KafkaProducer.java

@@ -9,6 +9,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 public class KafkaProducer {
+
     private final KafkaTemplate<String, String> kafkaTemplate;
 
     @Autowired

src/main/java/com/nisum/webfluxmongodbordermanagement/config/KafkaProducerConfig.java

@@ -18,6 +18,7 @@ public class KafkaProducerConfig {
 
     @Bean
     ProducerFactory<String, String> producerFactory() {
+
         Map<String, Object> config = new HashMap<>();
         config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
         config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

src/main/java/com/nisum/webfluxmongodbordermanagement/service/OrdersService.java

@@ -28,6 +28,7 @@ public class OrdersService {
     private String topicName;
 
     public Flux<Order> fetchOrdersAndPublish(){
+
         Flux<Order> orders =   orderRepository.findAll();
 
         Mono<List<Order>> ordersMono  = orders.collectList()

src/main/java/com/nisum/webfluxmongodbordermanagement/service/UserOrdersService.java

@@ -18,6 +18,7 @@ public class UserOrdersService {
      ReactiveMongoTemplate reactiveMongoTemplate;
 
     public Flux<UserOrders> getAllUsersWithOrders() {
+
         LookupOperation lookupOperation = LookupOperation.newLookup().
                 from("orders").
                 localField("userId").

src/main/java/com/nisum/webfluxmongodbordermanagement/service/UserService.java

@@ -14,6 +14,7 @@ public class UserService {
     private UserRepository userRepository;
 
     public Flux<User> getUsers(){
+
         return userRepository.findAll();
     }
 

src/main/resources/application.properties

 server.port=9090
 spring.data.mongodb.host=localhost
 spring.data.mongodb.port=27017
+
 spring.data.mongodb.database=user-orders-db
 order_topic=order_topic