added subscribe calls to ensure proper handling of asynchronous operations,...

added subscribe calls to ensure proper handling of asynchronous operations, warehouse updates successfully persisted for OMS

src/main/java/com/afp/ordermanagement/reactivekafkaservice/Receiver.java

@@ -56,18 +56,18 @@ public class Receiver {
         try {
             ObjectMapper objectMapper = new ObjectMapper();
             WarehouseOrder order = objectMapper.readValue(orderStr, WarehouseOrder.class);
-            log.info("ORDER objectMapper {}", order);
+            log.info("order received from warehouse {}", order);
             String orderId = order.getOrderId();
 
             System.out.println("About to try sending an email.");
             Mono<Order> convertedOrder = orderService.getOrderById(order.getOrderId());
-            convertedOrder.flatMap(o -> {
+            convertedOrder.doOnSuccess(o -> {
+                log.info("Existing order in the db: {}", o);
                 o.setOrderStatus(OrderStatus.valueOf(order.getStatus()));
                 emailService.emailCreator(o);
-                orderService.updateOrderByOrderId(orderId, o);
+                orderService.updateOrderByOrderId(orderId, o).subscribe();
                 System.out.println("ayayayayayayayaya receiver" + o);
-                return Mono.just(o);
-            }).block();
+            }).subscribe();
 
         } catch (Exception e) {
             log.error("Caught error on UpdateOrderStatus method", e);