Outbox Message and Scheduler class implemented part - 4.

order-service/order-container/pom.xml

@@ -36,6 +36,11 @@
             <artifactId>order-messaging</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>

order-service/order-container/src/test/java/com/food/order/system/OrderPaymentSagaTest.java

@@ -0,0 +1,121 @@
+package com.food.order.system;
+
+
+import com.food.order.system.data.access.outbox.payment.entity.PaymentOutboxEntity;
+import com.food.order.system.data.access.outbox.payment.repository.PaymentOutboxJpaRepository;
+import com.food.order.system.dto.message.PaymentResponse;
+import com.food.order.system.saga.OrderPaymentSaga;
+import com.food.order.system.saga.SagaStatus;
+import com.food.order.system.valueobject.PaymentStatus;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.test.context.jdbc.Sql;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+
+import static com.food.order.system.outbox.order.SagaConst.ORDER_PROCESSING_SAGA;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.springframework.test.context.jdbc.Sql.ExecutionPhase.AFTER_TEST_METHOD;
+
+@Slf4j
+@SpringBootTest(classes = OrderServiceApplication.class)
+@Sql(value = {"classpath:sql/OrderPaymentSagaTestSetUp.sql"})
+@Sql(value = {"classpath:sql/OrderPaymentSagaTestCleanUp.sql"}, executionPhase = AFTER_TEST_METHOD)
+class OrderPaymentSagaTest {
+
+    @Autowired
+    private OrderPaymentSaga orderPaymentSaga;
+
+    @Autowired
+    private PaymentOutboxJpaRepository paymentOutboxJpaRepository;
+
+    private final UUID SAGA_ID = UUID.fromString("15a497c1-0f4b-4eff-b9f4-c402c8c07afa");
+    private final UUID ORDER_ID = UUID.fromString("d215b5f8-0249-4dc5-89a3-51fd148cfb17");
+    private final UUID CUSTOMER_ID = UUID.fromString("d215b5f8-0249-4dc5-89a3-51fd148cfb41");
+    private final UUID PAYMENT_ID = UUID.randomUUID();
+    private final BigDecimal PRICE = new BigDecimal("100");
+
+    @Test
+    void testDoublePayment() {
+        orderPaymentSaga.process(getPaymentResponse());
+        orderPaymentSaga.process(getPaymentResponse());
+    }
+
+    @Test
+    void testDoublePaymentWithThreads() throws InterruptedException {
+        Thread thread1 = new Thread(() -> orderPaymentSaga.process(getPaymentResponse()));
+        Thread thread2 = new Thread(() -> orderPaymentSaga.process(getPaymentResponse()));
+
+        thread1.start();
+        thread2.start();
+
+        thread1.join();
+        thread2.join();
+
+        assertPaymentOutbox();
+    }
+
+    @Test
+    void testDoublePaymentWithLatch() throws InterruptedException {
+        CountDownLatch latch = new CountDownLatch(2);
+
+        Thread thread1 = new Thread(() -> {
+            try {
+                orderPaymentSaga.process(getPaymentResponse());
+            } catch (OptimisticLockingFailureException e) {
+                log.error("OptimisticLockingFailureException occurred for thread1");
+            } finally {
+                latch.countDown();
+            }
+        });
+
+        Thread thread2 = new Thread(() -> {
+            try {
+                orderPaymentSaga.process(getPaymentResponse());
+            } catch (OptimisticLockingFailureException e) {
+                log.error("OptimisticLockingFailureException occurred for thread2");
+            } finally {
+                latch.countDown();
+            }
+        });
+
+        thread1.start();
+        thread2.start();
+
+        latch.await();
+
+        assertPaymentOutbox();
+
+    }
+
+    private void assertPaymentOutbox() {
+        Optional<PaymentOutboxEntity> paymentOutboxEntity =
+                paymentOutboxJpaRepository.findByTypeAndSagaIdAndSagaStatusIn(ORDER_PROCESSING_SAGA, SAGA_ID,
+                        List.of(SagaStatus.PROCESSING));
+        assertTrue(paymentOutboxEntity.isPresent());
+    }
+
+    private PaymentResponse getPaymentResponse() {
+        return PaymentResponse.builder()
+                .id(UUID.randomUUID().toString())
+                .sagaId(SAGA_ID.toString())
+                .paymentStatus(PaymentStatus.COMPLETED)
+                .paymentId(PAYMENT_ID.toString())
+                .orderId(ORDER_ID.toString())
+                .customerId(CUSTOMER_ID.toString())
+                .price(PRICE)
+                .createdAt(Instant.now())
+                .failureMessages(new ArrayList<>())
+                .build();
+    }
+
+}

order-service/order-container/src/test/resources/sql/OrderPaymentSagaTestCleanUp.sql

@@ -0,0 +1,3 @@
+delete from "order".orders where id = 'd215b5f8-0249-4dc5-89a3-51fd148cfb17';
+
+delete from "order".payment_outbox where id = '8904808e-286f-449b-9b56-b63ba8351cf2';

order-service/order-container/src/test/resources/sql/OrderPaymentSagaTestSetUp.sql

@@ -0,0 +1,15 @@
+insert into "order".orders(id, customer_id, restaurant_id, tracking_id, price, order_status, failure_messages)
+values('d215b5f8-0249-4dc5-89a3-51fd148cfb17', 'd215b5f8-0249-4dc5-89a3-51fd148cfb41', 'd215b5f8-0249-4dc5-89a3-51fd148cfb45',
+ 'd215b5f8-0249-4dc5-89a3-51fd148cfb18', 100.00, 'PENDING', '');
+
+insert into "order".order_items(id, order_id, product_id, price, quantity, sub_total)
+values(1, 'd215b5f8-0249-4dc5-89a3-51fd148cfb17', 'd215b5f8-0249-4dc5-89a3-51fd148cfb47', 100.00, 1, 100.00);
+
+insert into "order".order_address(id, order_id, street, postal_code, city)
+values('d215b5f8-0249-4dc5-89a3-51fd148cfb15', 'd215b5f8-0249-4dc5-89a3-51fd148cfb17', 'test street', '1000AA', 'test city');
+
+insert into "order".payment_outbox(id, saga_id, created_at, type, payload, outbox_status, saga_status, order_status, version)
+values ('8904808e-286f-449b-9b56-b63ba8351cf2', '15a497c1-0f4b-4eff-b9f4-c402c8c07afa', current_timestamp, 'OrderProcessingSaga',
+ '{"price": 100, "orderId": "ef471dac-ec22-43a7-a3f4-9d04195567a5", "createdAt": "2022-01-07T16:21:42.917756+01:00",
+  "customerId": "d215b5f8-0249-4dc5-89a3-51fd148cfb41", "paymentOrderStatus": "PENDING"}',
+'STARTED', 'STARTED', 'PENDING', 0);

order-service/order-data-access/src/main/java/com/food/order/system/data/access/order/adapter/OrderRepositoryImpl.java

@@ -1,16 +1,16 @@
 package com.food.order.system.data.access.order.adapter;
 
-import com.food.order.system.ports.output.repository.OrderRepository;
 import com.food.order.system.data.access.order.mapper.OrderDataAccessMapper;
 import com.food.order.system.data.access.order.repository.OrderJpaRepository;
 import com.food.order.system.domain.entity.Order;
 import com.food.order.system.domain.valueobject.TrackingId;
+import com.food.order.system.ports.output.repository.OrderRepository;
+import com.food.order.system.valueobject.OrderId;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
 import java.util.Optional;
-import java.util.UUID;
 
 @Service
 @RequiredArgsConstructor
@@ -28,8 +28,8 @@ public class OrderRepositoryImpl implements OrderRepository {
     }
 
     @Override
-    public Optional<Order> findById(String orderId) {
-        return orderJpaRepository.findById(UUID.fromString(orderId))
+    public Optional<Order> findById(OrderId orderId) {
+        return orderJpaRepository.findById(orderId.getValue())
                 .map(orderDataAccessMapper::orderEntityToOrder);
     }
 

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/helper/OrderSagaHelper.java

@@ -4,11 +4,14 @@ import com.food.order.system.domain.entity.Order;
 import com.food.order.system.domain.exception.OrderNotFoundException;
 import com.food.order.system.ports.output.repository.OrderRepository;
 import com.food.order.system.saga.SagaStatus;
+import com.food.order.system.valueobject.OrderId;
 import com.food.order.system.valueobject.OrderStatus;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
+import java.util.UUID;
+
 @Slf4j
 @Component
 @RequiredArgsConstructor
@@ -17,7 +20,7 @@ public class OrderSagaHelper {
     private final OrderRepository orderRepository;
 
     public Order findOrder(String orderId) {
-        return orderRepository.findById(orderId)
+        return orderRepository.findById(new OrderId(UUID.fromString(orderId)))
                 .orElseThrow(() -> new OrderNotFoundException("Order not found -> Order id :" + orderId));
     }
 

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/ports/output/repository/OrderRepository.java

@@ -2,6 +2,7 @@ package com.food.order.system.ports.output.repository;
 
 import com.food.order.system.domain.entity.Order;
 import com.food.order.system.domain.valueobject.TrackingId;
+import com.food.order.system.valueobject.OrderId;
 
 import java.util.Optional;
 
@@ -9,7 +10,7 @@ public interface OrderRepository {
 
     Order save(Order order);
 
-    Optional<Order> findById(String trackingId);
+    Optional<Order> findById(OrderId trackingId);
 
     Optional<Order> findByTrackingId(TrackingId trackingId);
 

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/saga/OrderApprovalSaga.java

@@ -4,13 +4,14 @@ import com.food.order.system.domain.entity.Order;
 import com.food.order.system.domain.event.OrderCancelledEvent;
 import com.food.order.system.domain.exception.OrderDomainException;
 import com.food.order.system.domain.service.OrderDomainService;
+import com.food.order.system.dto.message.RestaurantApprovalResponse;
 import com.food.order.system.helper.OrderSagaHelper;
 import com.food.order.system.mapper.OrderDataMapper;
 import com.food.order.system.outbox.OutboxStatus;
 import com.food.order.system.outbox.model.approval.OrderApprovalOutboxMessage;
+import com.food.order.system.outbox.model.payment.OrderPaymentOutboxMessage;
 import com.food.order.system.outbox.scheduler.approval.ApprovalOutboxHelper;
 import com.food.order.system.outbox.scheduler.payment.PaymentOutboxHelper;
-import com.food.order.system.dto.message.RestaurantApprovalResponse;
 import com.food.order.system.valueobject.OrderStatus;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -19,6 +20,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
+import java.util.Optional;
 import java.util.UUID;
 
 import static com.food.order.system.DomainConstants.UTC;
@@ -35,107 +37,96 @@ public class OrderApprovalSaga implements SagaStep<RestaurantApprovalResponse> {
     private final ApprovalOutboxHelper approvalOutboxHelper;
 
 
-
     @Override
     @Transactional
-    public void process(RestaurantApprovalResponse data) {
-
-        var messageResponse =
+    public void process(RestaurantApprovalResponse restaurantApprovalResponse) {
+        var orderApprovalOutboxMessageResponse =
                 approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
-                        UUID.fromString(data.getSagaId()),
-                        SagaStatus.PROCESSING)
-                        .orElseThrow(() -> {
-                            log.error("Approval outbox message not found for saga id: {}", data.getSagaId());
-                                return new OrderDomainException("Approval outbox message not found for saga id: " + data.getSagaId());
-                        });
+                        UUID.fromString(restaurantApprovalResponse.getSagaId()),
+                        SagaStatus.PROCESSING).orElseThrow(
+                        () -> new OrderDomainException("OrderApprovalSaga: Order approval outbox message not found"));
 
-        var order = approveOrder(data);
+        var order = approveOrder(restaurantApprovalResponse);
 
         var sagaStatus = orderSagaHelper.orderStatusToSagaStatus(order.getStatus());
 
-        approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(messageResponse,order.getStatus(), sagaStatus));
-
-        approvalOutboxHelper.save(getUpdatedPaymentOutboxMessage(messageResponse.getSagaId(),
+        approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(orderApprovalOutboxMessageResponse,
                 order.getStatus(), sagaStatus));
 
+        paymentOutboxHelper.save(getUpdatedPaymentOutboxMessage(restaurantApprovalResponse.getSagaId(),
+                order.getStatus(), sagaStatus));
 
-
-        log.info("Order approved: {}", order);
+        log.info("Order with id: {} is approved", order.getId().getValue());
     }
 
-    private OrderApprovalOutboxMessage getUpdatedPaymentOutboxMessage(UUID sagaId,
-                                                                      OrderStatus status,
-                                                                      SagaStatus sagaStatus) {
-        var message =  approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
-                sagaId,
-                SagaStatus.PROCESSING)
-                .orElseThrow(() -> {
-                    log.error("Approval outbox message not found for saga id: {}", sagaId);
-                    return new OrderDomainException("Approval outbox message not found for saga id: " + sagaId);
-                });
+    @Override
+    @Transactional
+    public void rollback(RestaurantApprovalResponse restaurantApprovalResponse) {
+        var orderApprovalOutboxMessageResponse =
+                approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
+                        UUID.fromString(restaurantApprovalResponse.getSagaId()),
+                        SagaStatus.PROCESSING).orElseThrow(
+                        () -> new OrderDomainException("OrderApprovalSaga: Order approval outbox message not found"));
 
-        message.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
-        message.setSagaStatus(sagaStatus);
-        message.setOrderStatus(status);
-        return message;
+        var domainEvent = rollbackOrder(restaurantApprovalResponse);
+
+        var sagaStatus = orderSagaHelper.orderStatusToSagaStatus(domainEvent.getOrder().getStatus());
+
+        approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(orderApprovalOutboxMessageResponse,
+                domainEvent.getOrder().getStatus(), sagaStatus));
+
+        paymentOutboxHelper.savePaymentOutboxMessage(orderDataMapper
+                        .orderCancelledEventToOrderPaymentEventPayload(domainEvent),
+                domainEvent.getOrder().getStatus(),
+                sagaStatus,
+                OutboxStatus.STARTED,
+                UUID.fromString(restaurantApprovalResponse.getSagaId()));
+
+        log.info("Order with id: {} is cancelling", domainEvent.getOrder().getId().getValue());
     }
 
-    private OrderApprovalOutboxMessage getUpdatedApprovalOutboxMessage(OrderApprovalOutboxMessage messageResponse,
-                                                                       OrderStatus status,
-                                                                       SagaStatus sagaStatus) {
-        messageResponse.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
-        messageResponse.setSagaStatus(sagaStatus);
-        messageResponse.setOrderStatus(status);
-        return messageResponse;
-    }
-
-
-    private Order approveOrder(RestaurantApprovalResponse data) {
-        var order = orderSagaHelper.findOrder(data.getOrderId());
+    private Order approveOrder(RestaurantApprovalResponse restaurantApprovalResponse) {
+        log.info("Approving order with id: {}", restaurantApprovalResponse.getOrderId());
+        Order order = orderSagaHelper.findOrder(restaurantApprovalResponse.getOrderId());
         orderDomainService.approve(order);
         orderSagaHelper.saveOrder(order);
         return order;
     }
 
-
-    @Override
-    @Transactional
-    public void rollback(RestaurantApprovalResponse data) {
-
-        var message = approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
-                UUID.fromString(data.getSagaId()),
-                SagaStatus.PROCESSING)
-                .orElseThrow(
-                        () -> {
-                            log.error("Approval outbox message not found for saga id: {}", data.getSagaId());
-                            return new OrderDomainException("Approval outbox message not found for saga id: " +
-                                    data.getSagaId());
-                        }
-                );
-
-        var event = rollbackOrder(data);
-
-        var sagaStatus = orderSagaHelper.orderStatusToSagaStatus(event.getOrder().getStatus());
-
-        approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(message, event.getOrder().getStatus(), sagaStatus));
-
-        paymentOutboxHelper.savePaymentOutboxMessage(
-                orderDataMapper.orderCancelledEventToOrderPaymentEventPayload(event),
-                event.getOrder().getStatus(),
-                sagaStatus,
-                OutboxStatus.STARTED,
-                message.getSagaId()
-        );
-
-        log.info("Order cancelled event id: {}", event.getOrder().getId());
+    private OrderApprovalOutboxMessage getUpdatedApprovalOutboxMessage(OrderApprovalOutboxMessage
+                                                                               orderApprovalOutboxMessage,
+                                                                       OrderStatus
+                                                                               orderStatus,
+                                                                       SagaStatus
+                                                                               sagaStatus) {
+        orderApprovalOutboxMessage.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
+        orderApprovalOutboxMessage.setOrderStatus(orderStatus);
+        orderApprovalOutboxMessage.setSagaStatus(sagaStatus);
+        return orderApprovalOutboxMessage;
     }
 
-    private OrderCancelledEvent rollbackOrder(RestaurantApprovalResponse data) {
-        var order = orderSagaHelper.findOrder(data.getOrderId());
-        var event = orderDomainService.cancelOrderPayment(
-                order,
-                data.getFailureMessages());
+    private OrderPaymentOutboxMessage getUpdatedPaymentOutboxMessage(String sagaId,
+                                                                     OrderStatus orderStatus,
+                                                                     SagaStatus sagaStatus) {
+        Optional<OrderPaymentOutboxMessage> orderPaymentOutboxMessageResponse = paymentOutboxHelper
+                .getPaymentOutboxMessageBySagaIdAndSagaStatus(UUID.fromString(sagaId), SagaStatus.PROCESSING);
+        if (orderPaymentOutboxMessageResponse.isEmpty()) {
+            throw new OrderDomainException("Payment outbox message cannot be found in " +
+                    SagaStatus.PROCESSING.name() + " state");
+        }
+        OrderPaymentOutboxMessage orderPaymentOutboxMessage = orderPaymentOutboxMessageResponse.get();
+        orderPaymentOutboxMessage.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
+        orderPaymentOutboxMessage.setOrderStatus(orderStatus);
+        orderPaymentOutboxMessage.setSagaStatus(sagaStatus);
+        return orderPaymentOutboxMessage;
+    }
+
+    private OrderCancelledEvent rollbackOrder(RestaurantApprovalResponse restaurantApprovalResponse) {
+        log.info("Cancelling order with id: {}", restaurantApprovalResponse.getOrderId());
+        Order order = orderSagaHelper.findOrder(restaurantApprovalResponse.getOrderId());
+        OrderCancelledEvent domainEvent = orderDomainService.cancelOrderPayment(order,
+                restaurantApprovalResponse.getFailureMessages());
         orderSagaHelper.saveOrder(order);
-        return event;
+        return domainEvent;
     }
 }

order-service/order-domain/order-application-service/src/main/java/com/food/order/system/saga/OrderPaymentSaga.java

@@ -3,15 +3,18 @@ package com.food.order.system.saga;
 import com.food.order.system.domain.entity.Order;
 import com.food.order.system.domain.event.OrderPaidEvent;
 import com.food.order.system.domain.exception.OrderDomainException;
+import com.food.order.system.domain.exception.OrderNotFoundException;
 import com.food.order.system.domain.service.OrderDomainService;
 import com.food.order.system.dto.message.PaymentResponse;
+import com.food.order.system.helper.OrderSagaHelper;
+import com.food.order.system.mapper.OrderDataMapper;
 import com.food.order.system.outbox.OutboxStatus;
 import com.food.order.system.outbox.model.approval.OrderApprovalOutboxMessage;
 import com.food.order.system.outbox.model.payment.OrderPaymentOutboxMessage;
-import com.food.order.system.helper.OrderSagaHelper;
-import com.food.order.system.mapper.OrderDataMapper;
 import com.food.order.system.outbox.scheduler.approval.ApprovalOutboxHelper;
 import com.food.order.system.outbox.scheduler.payment.PaymentOutboxHelper;
+import com.food.order.system.ports.output.repository.OrderRepository;
+import com.food.order.system.valueobject.OrderId;
 import com.food.order.system.valueobject.OrderStatus;
 import com.food.order.system.valueobject.PaymentStatus;
 import lombok.RequiredArgsConstructor;
@@ -21,6 +24,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
+import java.util.Optional;
 import java.util.UUID;
 
 import static com.food.order.system.DomainConstants.UTC;
@@ -29,112 +33,103 @@ import static com.food.order.system.DomainConstants.UTC;
 @Component
 @RequiredArgsConstructor
 public class OrderPaymentSaga implements SagaStep<PaymentResponse> {
-
     private final OrderDomainService orderDomainService;
+    private final OrderRepository orderRepository;
+    private final PaymentOutboxHelper paymentOutboxHelper;
+    private final ApprovalOutboxHelper approvalOutboxHelper;
     private final OrderSagaHelper orderSagaHelper;
     private final OrderDataMapper orderDataMapper;
-    private final ApprovalOutboxHelper approvalOutboxHelper;
-    private final PaymentOutboxHelper paymentOutboxHelper;
-
 
     @Override
     @Transactional
-    public void process(PaymentResponse data) {
-
-        var messageResponse =
+    public void process(PaymentResponse paymentResponse) {
+        Optional<OrderPaymentOutboxMessage> orderPaymentOutboxMessageResponse =
                 paymentOutboxHelper.getPaymentOutboxMessageBySagaIdAndSagaStatus(
-                UUID.fromString(data.getSagaId()),
-                SagaStatus.STARTED)
-                        .orElseThrow(() -> {
-                            log.error("Payment outbox message not found for saga id: {}", data.getSagaId());
-                            return new OrderDomainException("Payment outbox message not found for saga id: " + data.getSagaId());
-                        });
+                        UUID.fromString(paymentResponse.getSagaId()),
+                        SagaStatus.STARTED);
 
+        if (orderPaymentOutboxMessageResponse.isEmpty()) {
+            log.info("An outbox message with saga id: {} is already processed!", paymentResponse.getSagaId());
+            return;
+        }
 
-        var paidEvent = completePaymentForOrder(data);
+        OrderPaymentOutboxMessage orderPaymentOutboxMessage = orderPaymentOutboxMessageResponse.get();
 
-        var sagaStatus = orderSagaHelper.orderStatusToSagaStatus(paidEvent.getOrder().getStatus());
+        OrderPaidEvent domainEvent = completePaymentForOrder(paymentResponse);
 
-        paymentOutboxHelper.save(getUpdatedPaymentOutboxMessage(messageResponse,
-                                                                paidEvent.getOrder().getStatus(),
-                                                                sagaStatus));
+        SagaStatus sagaStatus = orderSagaHelper.orderStatusToSagaStatus(domainEvent.getOrder().getStatus());
 
-        approvalOutboxHelper.saveApprovalOutboxMessage(
-                orderDataMapper.orderPaidEventToOrderApprovalEventPayload(paidEvent),
-                paidEvent.getOrder().getStatus(),
+        paymentOutboxHelper.save(getUpdatedPaymentOutboxMessage(orderPaymentOutboxMessage,
+                domainEvent.getOrder().getStatus(), sagaStatus));
+
+        approvalOutboxHelper
+                .saveApprovalOutboxMessage(orderDataMapper.orderPaidEventToOrderApprovalEventPayload(domainEvent),
+                        domainEvent.getOrder().getStatus(),
                         sagaStatus,
                         OutboxStatus.STARTED,
-                messageResponse.getSagaId()
-                );
+                        UUID.fromString(paymentResponse.getSagaId()));
 
-        log.info("Payment completed for order with id: {}", paidEvent.getOrder().getId().getValue());
-    }
-
-    private OrderPaymentOutboxMessage getUpdatedPaymentOutboxMessage(OrderPaymentOutboxMessage messageResponse,
-                                                                     OrderStatus status,
-                                                                     SagaStatus sagaStatus) {
-        messageResponse.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
-        messageResponse.setOrderStatus(status);
-        messageResponse.setSagaStatus(sagaStatus);
-        return messageResponse;
+        log.info("Order with id: {} is paid", domainEvent.getOrder().getId().getValue());
     }
 
     @Override
     @Transactional
-    public void rollback(PaymentResponse data) {
+    public void rollback(PaymentResponse paymentResponse) {
 
-        var messageResponse =
+        Optional<OrderPaymentOutboxMessage> orderPaymentOutboxMessageResponse =
                 paymentOutboxHelper.getPaymentOutboxMessageBySagaIdAndSagaStatus(
-                        UUID.fromString(data.getSagaId()),
-                        getCurrentSagaStatus(data.getPaymentStatus()))
-                        .orElseThrow(
-                                () -> {
-                                    log.error("Payment outbox message not found for saga id: {}", data.getSagaId());
-                                    return new OrderDomainException("Payment outbox message not found for saga id: " + data.getSagaId());
-                                }
-                        );
+                        UUID.fromString(paymentResponse.getSagaId()),
+                        getCurrentSagaStatus(paymentResponse.getPaymentStatus()));
 
-        var orderRollback = rollbackPaymentForOrder(data);
-
-        var sagaStatus = orderSagaHelper.orderStatusToSagaStatus(orderRollback.getStatus());
-
-        paymentOutboxHelper.save(getUpdatedPaymentOutboxMessage(messageResponse,
-                                                                orderRollback.getStatus(),
-                                                                sagaStatus));
-
-        if (data.getPaymentStatus().equals(PaymentStatus.CANCELED)) {
-            approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(data.getSagaId(),
-                    orderRollback.getStatus(),
-                    sagaStatus));
+        if (orderPaymentOutboxMessageResponse.isEmpty()) {
+            log.info("An outbox message with saga id: {} is already roll backed!", paymentResponse.getSagaId());
+            return;
         }
 
-        log.info("Payment rolled back for order with id: {}", orderRollback.getId());
+        OrderPaymentOutboxMessage orderPaymentOutboxMessage = orderPaymentOutboxMessageResponse.get();
+
+        Order order = rollbackPaymentForOrder(paymentResponse);
+
+        SagaStatus sagaStatus = orderSagaHelper.orderStatusToSagaStatus(order.getStatus());
+
+        paymentOutboxHelper.save(getUpdatedPaymentOutboxMessage(orderPaymentOutboxMessage,
+                order.getStatus(), sagaStatus));
+
+        if (paymentResponse.getPaymentStatus() == PaymentStatus.CANCELED) {
+            approvalOutboxHelper.save(getUpdatedApprovalOutboxMessage(paymentResponse.getSagaId(),
+                    order.getStatus(), sagaStatus));
         }
 
-    private OrderApprovalOutboxMessage getUpdatedApprovalOutboxMessage(String sagaId,
-                                                                       OrderStatus orderStatus,
-                                                                       SagaStatus sagaStatus) {
-        var orderApprovalOutboxMessageResponse =
-                approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
-                        UUID.fromString(sagaId),
-                        SagaStatus.COMPENSATING);
-        if (orderApprovalOutboxMessageResponse.isEmpty()) {
-            throw new OrderDomainException("Approval outbox message could not be found in " +
-                    SagaStatus.COMPENSATING.name() + " status!");
-        }
-        var orderApprovalOutboxMessage = orderApprovalOutboxMessageResponse.get();
-        orderApprovalOutboxMessage.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
-        orderApprovalOutboxMessage.setOrderStatus(orderStatus);
-        orderApprovalOutboxMessage.setSagaStatus(sagaStatus);
-        return orderApprovalOutboxMessage;
+        log.info("Order with id: {} is cancelled", order.getId().getValue());
     }
 
-    private Order rollbackPaymentForOrder(PaymentResponse paymentResponse) {
-        log.info("Cancelling order with id: {}", paymentResponse.getOrderId());
-        var order = orderSagaHelper.findOrder(paymentResponse.getOrderId());
-        orderDomainService.cancelOrder(order, paymentResponse.getFailureMessages());
-        orderSagaHelper.saveOrder(order);
-        return order;
+    private Order findOrder(String orderId) {
+        Optional<Order> orderResponse = orderRepository.findById(new OrderId(UUID.fromString(orderId)));
+        if (orderResponse.isEmpty()) {
+            log.error("Order with id: {} could not be found!", orderId);
+            throw new OrderNotFoundException("Order with id " + orderId + " could not be found!");
+        }
+        return orderResponse.get();
+    }
+
+    private OrderPaymentOutboxMessage getUpdatedPaymentOutboxMessage(OrderPaymentOutboxMessage
+                                                                             orderPaymentOutboxMessage,
+                                                                     OrderStatus
+                                                                             orderStatus,
+                                                                     SagaStatus
+                                                                             sagaStatus) {
+        orderPaymentOutboxMessage.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
+        orderPaymentOutboxMessage.setOrderStatus(orderStatus);
+        orderPaymentOutboxMessage.setSagaStatus(sagaStatus);
+        return orderPaymentOutboxMessage;
+    }
+
+    private OrderPaidEvent completePaymentForOrder(PaymentResponse paymentResponse) {
+        log.info("Completing payment for order with id: {}", paymentResponse.getOrderId());
+        Order order = findOrder(paymentResponse.getOrderId());
+        OrderPaidEvent domainEvent = orderDomainService.payOrder(order);
+        orderRepository.save(order);
+        return domainEvent;
     }
 
     private SagaStatus[] getCurrentSagaStatus(PaymentStatus paymentStatus) {
@@ -145,11 +140,29 @@ public class OrderPaymentSaga implements SagaStep<PaymentResponse> {
         };
     }
 
-    private OrderPaidEvent completePaymentForOrder(PaymentResponse data) {
-        var order = orderSagaHelper.findOrder(data.getOrderId());
-        var paidEvent = orderDomainService.payOrder(order);
-        orderSagaHelper.saveOrder(order);
-        log.info("Payment completed for order with id: {}", order.getId());
-        return paidEvent;
+    private Order rollbackPaymentForOrder(PaymentResponse paymentResponse) {
+        log.info("Cancelling order with id: {}", paymentResponse.getOrderId());
+        Order order = findOrder(paymentResponse.getOrderId());
+        orderDomainService.cancelOrder(order, paymentResponse.getFailureMessages());
+        orderRepository.save(order);
+        return order;
+    }
+
+    private OrderApprovalOutboxMessage getUpdatedApprovalOutboxMessage(String sagaId,
+                                                                       OrderStatus orderStatus,
+                                                                       SagaStatus sagaStatus) {
+        Optional<OrderApprovalOutboxMessage> orderApprovalOutboxMessageResponse =
+                approvalOutboxHelper.getApprovalOutboxMessageBySagaIdAndSagaStatus(
+                        UUID.fromString(sagaId),
+                        SagaStatus.COMPENSATING);
+        if (orderApprovalOutboxMessageResponse.isEmpty()) {
+            throw new OrderDomainException("Approval outbox message could not be found in " +
+                    SagaStatus.COMPENSATING.name() + " status!");
+        }
+        OrderApprovalOutboxMessage orderApprovalOutboxMessage = orderApprovalOutboxMessageResponse.get();
+        orderApprovalOutboxMessage.setProcessedAt(ZonedDateTime.now(ZoneId.of(UTC)));
+        orderApprovalOutboxMessage.setOrderStatus(orderStatus);
+        orderApprovalOutboxMessage.setSagaStatus(sagaStatus);
+        return orderApprovalOutboxMessage;
     }
 }

order-service/order-domain/order-application-service/src/test/java/com/food/order/system/OrderApplicationServiceTest.java

@@ -176,7 +176,7 @@ class OrderApplicationServiceTest {
     void testCreateOrderWithWrongTotalPrice() {
        OrderDomainException orderDomainException = assertThrows(OrderDomainException.class,
                 () -> orderApplicationService.createOrder(createOrderCommandWrongPrice));
-       assertEquals("Total price: 250.00 is not equal to Order items total: 200.00!",
+       assertEquals("Order total price is not equal to the sum of order items prices",
                orderDomainException.getMessage());
     }
 
@@ -185,7 +185,7 @@ class OrderApplicationServiceTest {
        OrderDomainException orderDomainException = assertThrows(OrderDomainException.class,
                 () -> orderApplicationService.createOrder(createOrderCommandWrongProductPrice));
        assertEquals(orderDomainException.getMessage(),
-               "Order item price: 60.00 is not valid for product " + PRODUCT_ID);
+               "Order item price is not valid");
     }
 
     @Test
@@ -201,7 +201,7 @@ class OrderApplicationServiceTest {
        OrderDomainException orderDomainException = assertThrows(OrderDomainException.class,
                () -> orderApplicationService.createOrder(createOrderCommand));
        assertEquals(orderDomainException.getMessage(),
-               "Restaurant with id " + RESTAURANT_ID + " is currently not active!");
+               "Restaurant is not active, please try again later. Restaurant id: " + restaurantResponse.getId());
     }
 
     private OrderPaymentOutboxMessage getOrderPaymentOutboxMessage() {

payment-service/payment-container/src/main/resources/application.yml

@@ -8,6 +8,8 @@ logging:
 payment-service:
   payment-request-topic-name: payment-request-value
   payment-response-topic-name: payment-response-value
+  outbox-scheduler-fixed-rate: 10000
+  outbox-scheduler-initial-delay: 10000
 
 spring:
   jpa:

payment-service/payment-container/src/main/resources/init-schema.sql

@@ -45,3 +45,30 @@ CREATE TABLE "payment".credit_history
     type transaction_type NOT NULL,
     CONSTRAINT credit_history_pkey PRIMARY KEY (id)
 );
+
+DROP TYPE IF EXISTS outbox_status;
+CREATE TYPE outbox_status AS ENUM ('STARTED', 'COMPLETED', 'FAILED');
+
+DROP TABLE IF EXISTS "payment".order_outbox CASCADE;
+
+CREATE TABLE "payment".order_outbox
+(
+    id uuid NOT NULL,
+    saga_id uuid NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    processed_at TIMESTAMP WITH TIME ZONE,
+    type character varying COLLATE pg_catalog."default" NOT NULL,
+    payload jsonb NOT NULL,
+    outbox_status outbox_status NOT NULL,
+    payment_status payment_status NOT NULL,
+    version integer NOT NULL,
+    CONSTRAINT order_outbox_pkey PRIMARY KEY (id)
+);
+
+CREATE INDEX "payment_order_outbox_saga_status"
+    ON "payment".order_outbox
+        (type, payment_status);
+
+CREATE UNIQUE INDEX "payment_order_outbox_saga_id_payment_status_outbox_status"
+    ON "payment".order_outbox
+        (type, saga_id, payment_status, outbox_status);

payment-service/payment-domain/payment-application-service/pom.xml

@@ -34,6 +34,25 @@
             <artifactId>payment-domain-core</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>outbox</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.food.order</groupId>
+            <artifactId>saga</artifactId>
+        </dependency>
+
     </dependencies>
 
 

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/PaymentRequestHelper.java

@@ -1,8 +1,11 @@
 package com.food.order.system.payment.application.service;
 
+import com.food.order.system.outbox.OutboxStatus;
 import com.food.order.system.payment.application.service.dto.PaymentRequest;
 import com.food.order.system.payment.application.service.exception.PaymentApplicationServiceException;
 import com.food.order.system.payment.application.service.mapper.PaymentDataMapper;
+import com.food.order.system.payment.application.service.outbox.scheduler.OrderOutboxHelper;
+import com.food.order.system.payment.application.service.ports.output.message.publisher.PaymentResponseMessagePublisher;
 import com.food.order.system.payment.application.service.ports.output.repository.CreditEntryRepository;
 import com.food.order.system.payment.application.service.ports.output.repository.CreditHistoryRepository;
 import com.food.order.system.payment.application.service.ports.output.repository.PaymentRepository;
@@ -10,12 +13,8 @@ import com.food.order.system.payment.service.domain.PaymentDomainService;
 import com.food.order.system.payment.service.domain.entity.CreditEntry;
 import com.food.order.system.payment.service.domain.entity.CreditHistory;
 import com.food.order.system.payment.service.domain.entity.Payment;
-import com.food.order.system.payment.service.domain.event.PaymentCancelledEvent;
-import com.food.order.system.payment.service.domain.event.PaymentCompletedEvent;
-import com.food.order.system.payment.service.domain.event.PaymentEvent;
-import com.food.order.system.payment.service.domain.event.PaymentFailedEvent;
-import com.food.order.system.event.publisher.DomainEventPublisher;
 import com.food.order.system.valueobject.CustomerId;
+import com.food.order.system.valueobject.PaymentStatus;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
@@ -30,18 +29,24 @@ import java.util.UUID;
 @RequiredArgsConstructor
 public class PaymentRequestHelper {
     private final PaymentDomainService paymentDomainService;
+
+    private final OrderOutboxHelper orderOutboxHelper;
+
+    private final PaymentResponseMessagePublisher paymentResponseMessagePublisher;
     private final PaymentDataMapper paymentDataMapper;
     private final PaymentRepository paymentRepository;
     private final CreditEntryRepository creditEntryRepository;
     private final CreditHistoryRepository creditHistoryRepository;
 
-    private final DomainEventPublisher<PaymentCompletedEvent> publisher;
-    private final DomainEventPublisher<PaymentCancelledEvent> publisherCancelled;
-
-    private final DomainEventPublisher<PaymentFailedEvent> failedEventDomainEventPublisher;
 
     @Transactional
-    public PaymentEvent persistPayment(PaymentRequest paymentRequest) {
+    public void persistPayment(PaymentRequest paymentRequest) {
+
+        if (publishIfOutboxMessageProcessedForPayment(paymentRequest,PaymentStatus.COMPLETED)) {
+            log.info("Outbox Message with sagaId : {} already save !", paymentRequest.getSagaId());
+            return;
+        }
+
         log.info("Received payment complete event for id : {}", paymentRequest.getOrderId());
         var payment = paymentDataMapper.paymentRequestModelToPayment(paymentRequest);
         var creditEntry = getCreditEntry(payment.getCustomerId());
@@ -49,14 +54,40 @@ public class PaymentRequestHelper {
         List<String> failureMessage = new ArrayList<>();
 
         var paymentEvent = paymentDomainService.validateAndInitializePayment
-                (payment, creditEntry, creditHistory, failureMessage,publisher,failedEventDomainEventPublisher);
+                (payment, creditEntry, creditHistory, failureMessage);
 
         persistDbObject(payment, creditEntry, creditHistory, failureMessage);
-        return paymentEvent;
+
+        orderOutboxHelper.saveOrderOutboxMessage(paymentDataMapper.paymentEventToOrderEventPayload(paymentEvent),
+                paymentEvent.getPayment().getStatus(),
+                OutboxStatus.STARTED,
+                UUID.fromString(paymentRequest.getSagaId()));
+
+
+
+
+    }
+
+    private boolean publishIfOutboxMessageProcessedForPayment(PaymentRequest paymentRequest,
+                                                              PaymentStatus paymentStatus) {
+        var outboxMessage = orderOutboxHelper.getCompletedOrderOutboxMessageBySagaIdAndPaymentStatus(
+                UUID.fromString(paymentRequest.getSagaId()), paymentStatus);
+        if (outboxMessage.isPresent()) {
+            paymentResponseMessagePublisher.publish(outboxMessage.get(),
+                    orderOutboxHelper::updateOutboxMessage);
+            return true;
+        }
+        return false;
     }
 
 
-    public PaymentEvent persistCancelPayment(PaymentRequest paymentRequest) {
+    public void persistCancelPayment(PaymentRequest paymentRequest) {
+
+        if (publishIfOutboxMessageProcessedForPayment(paymentRequest,PaymentStatus.CANCELED)) {
+            log.info("Outbox Message with sagaId : {} already save !", paymentRequest.getSagaId());
+            return;
+        }
+
         log.info("Received payment cancel event for id : {}", paymentRequest.getOrderId());
         var payment = paymentRepository.findByOrderId
                 (UUID.fromString(paymentRequest.getOrderId())).orElseThrow(
@@ -65,13 +96,22 @@ public class PaymentRequestHelper {
         var creditHistory = getCreditHistory(payment.getCustomerId());
         List<String> failureMessage = new ArrayList<>();
         var paymentEvent = paymentDomainService.validateAndCancelledPayment
-                (payment, creditEntry, creditHistory, failureMessage,publisherCancelled,failedEventDomainEventPublisher);
+                (payment, creditEntry, creditHistory, failureMessage);
 
         persistDbObject(payment, creditEntry, creditHistory, failureMessage);
-        return paymentEvent;
+
+
+        orderOutboxHelper.saveOrderOutboxMessage(paymentDataMapper.paymentEventToOrderEventPayload(paymentEvent),
+                paymentEvent.getPayment().getStatus(),
+                OutboxStatus.STARTED,
+                UUID.fromString(paymentRequest.getSagaId()));
+
     }
 
-    private void persistDbObject(Payment payment, CreditEntry creditEntry, List<CreditHistory> creditHistory, List<String> failureMessage) {
+    private void persistDbObject(Payment payment,
+                                 CreditEntry creditEntry,
+                                 List<CreditHistory> creditHistory,
+                                 List<String> failureMessage) {
         paymentRepository.save(payment);
         if (failureMessage.isEmpty()) {
             creditEntryRepository.save(creditEntry);
@@ -80,11 +120,13 @@ public class PaymentRequestHelper {
     }
     private List<CreditHistory> getCreditHistory(CustomerId customerId) {
         return creditHistoryRepository.findByCustomerId(customerId).orElseThrow(
-                () -> new PaymentApplicationServiceException("No credit history found for customer id : " + customerId));
+                () -> new PaymentApplicationServiceException
+                        ("No credit history found for customer id : " + customerId));
     }
 
     private CreditEntry getCreditEntry(CustomerId customerId) {
         return creditEntryRepository.findByCustomerId(customerId).orElseThrow(
-                () -> new PaymentApplicationServiceException("Credit entry not found for customer id : " + customerId));
+                () -> new PaymentApplicationServiceException
+                        ("Credit entry not found for customer id : " + customerId));
     }
 }

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/PaymentRequestMessageListenerImpl.java

@@ -2,7 +2,6 @@ package com.food.order.system.payment.application.service;
 
 import com.food.order.system.payment.application.service.dto.PaymentRequest;
 import com.food.order.system.payment.application.service.ports.input.message.listener.PaymentRequestMessageListener;
-import com.food.order.system.payment.service.domain.event.PaymentEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -16,20 +15,13 @@ public class PaymentRequestMessageListenerImpl implements PaymentRequestMessageL
 
     @Override
     public void completePayment(PaymentRequest paymentRequest) {
-        var event = paymentRequestHelper.persistPayment(paymentRequest);
-        fireEvent(event);
+        paymentRequestHelper.persistPayment(paymentRequest);
     }
     @Override
     public void cancelPayment(PaymentRequest paymentRequest) {
-        var event = paymentRequestHelper.persistCancelPayment(paymentRequest);
-        fireEvent(event);
-    }
-    private void fireEvent(PaymentEvent event) {
-
-        log.info("Firing event : {}", event);
-        event.fire();
-
+       paymentRequestHelper.persistCancelPayment(paymentRequest);
     }
 
 
+
 }

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/exception/PaymentDomainException.java

@@ -0,0 +1,14 @@
+package com.food.order.system.payment.application.service.exception;
+
+import com.food.order.system.exception.DomainException;
+
+public class PaymentDomainException extends DomainException {
+
+    public PaymentDomainException(String message) {
+        super(message);
+    }
+
+    public PaymentDomainException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/mapper/PaymentDataMapper.java

@@ -1,7 +1,9 @@
 package com.food.order.system.payment.application.service.mapper;
 
 import com.food.order.system.payment.application.service.dto.PaymentRequest;
+import com.food.order.system.payment.application.service.outbox.model.OrderEventPayload;
 import com.food.order.system.payment.service.domain.entity.Payment;
+import com.food.order.system.payment.service.domain.event.PaymentEvent;
 import com.food.order.system.valueobject.CustomerId;
 import com.food.order.system.valueobject.Money;
 import com.food.order.system.valueobject.OrderId;
@@ -20,4 +22,17 @@ public class PaymentDataMapper {
                 .price(new Money(paymentRequest.getPrice()))
                 .build();
     }
+
+    public OrderEventPayload paymentEventToOrderEventPayload(PaymentEvent payment) {
+        return OrderEventPayload.builder()
+                .orderId(payment.getPayment().getOrderId().getValue().toString())
+                .customerId(payment.getPayment().getCustomerId().getValue().toString())
+                .price(payment.getPayment().getPrice().getAmount())
+                .paymentId(payment.getPayment().getId().toString())
+                .createdAt(payment.getCreatedAt())
+                .failureMessages(payment.getFailureMessages())
+                .paymentStatus(payment.getPayment().getStatus().toString())
+                .build();
+
+    }
 }

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/outbox/model/OrderEventPayload.java

@@ -0,0 +1,38 @@
+package com.food.order.system.payment.application.service.outbox.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+
+import java.math.BigDecimal;
+import java.time.ZonedDateTime;
+import java.util.List;
+
+@Getter
+@Builder
+@AllArgsConstructor
+public class OrderEventPayload {
+
+    @JsonProperty
+    private String paymentId;
+
+    @JsonProperty
+    private String customerId;
+
+    @JsonProperty
+    private String orderId;
+
+    @JsonProperty
+    private BigDecimal price;
+
+    @JsonProperty
+    private String paymentStatus;
+
+    @JsonProperty
+    private ZonedDateTime createdAt;
+
+    @JsonProperty
+    private List<String> failureMessages;
+
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/outbox/model/OrderOutboxMessage.java

@@ -0,0 +1,38 @@
+package com.food.order.system.payment.application.service.outbox.model;
+
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.valueobject.PaymentStatus;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.time.ZonedDateTime;
+import java.util.UUID;
+
+@Getter
+@Builder
+@AllArgsConstructor
+public class OrderOutboxMessage {
+
+    private UUID id;
+
+    private UUID sagaId;
+
+    private ZonedDateTime createdAt;
+
+    private ZonedDateTime processedAt;
+
+    private String type;
+
+    private String payload;
+
+    private PaymentStatus paymentStatus;
+
+    @Setter
+    private OutboxStatus outboxStatus;
+
+    private int version;
+
+
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/outbox/scheduler/OrderOutboxCleanerScheduler.java

@@ -0,0 +1,36 @@
+package com.food.order.system.payment.application.service.outbox.scheduler;
+
+
+import com.food.order.system.outbox.OutboxScheduler;
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.payment.application.service.outbox.model.OrderOutboxMessage;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.Optional;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OrderOutboxCleanerScheduler implements OutboxScheduler {
+
+    private final OrderOutboxHelper orderOutboxHelper;
+
+    @Override
+    @Transactional
+    @Scheduled(cron = "@midnight")
+    public void processOutboxMessage() {
+        Optional<List<OrderOutboxMessage>> outboxMessagesResponse =
+                orderOutboxHelper.getOrderOutboxMessageByOutboxStatus(OutboxStatus.COMPLETED);
+        if (outboxMessagesResponse.isPresent() && outboxMessagesResponse.get().size() > 0) {
+            List<OrderOutboxMessage> outboxMessages = outboxMessagesResponse.get();
+            log.info("Received {} OrderOutboxMessage for clean-up!", outboxMessages.size());
+            orderOutboxHelper.deleteOrderOutboxMessageByOutboxStatus(OutboxStatus.COMPLETED);
+            log.info("Deleted {} OrderOutboxMessage!", outboxMessages.size());
+        }
+    }
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/outbox/scheduler/OrderOutboxHelper.java

@@ -0,0 +1,96 @@
+package com.food.order.system.payment.application.service.outbox.scheduler;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.payment.application.service.outbox.model.OrderEventPayload;
+import com.food.order.system.payment.application.service.outbox.model.OrderOutboxMessage;
+import com.food.order.system.payment.application.service.ports.output.repository.OrderOutboxRepository;
+import com.food.order.system.payment.application.service.exception.PaymentDomainException;
+import com.food.order.system.valueobject.PaymentStatus;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.UUID;
+
+import static com.food.order.system.DomainConstants.UTC;
+import static com.food.order.system.outbox.order.SagaConst.ORDER_PROCESSING_SAGA;
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class OrderOutboxHelper {
+
+    private final ObjectMapper objectMapper;
+    private final OrderOutboxRepository orderOutboxRepository;
+
+
+    @Transactional(readOnly = true)
+    public Optional<OrderOutboxMessage> getCompletedOrderOutboxMessageBySagaIdAndPaymentStatus(UUID sagaId,
+                                                                                               PaymentStatus
+                                                                                                       paymentStatus) {
+        return orderOutboxRepository.findByTypeAndSagaIdAndPaymentStatusAndOutboxStatus(ORDER_PROCESSING_SAGA, sagaId,
+                paymentStatus, OutboxStatus.COMPLETED);
+    }
+
+    @Transactional(readOnly = true)
+    public Optional<List<OrderOutboxMessage>> getOrderOutboxMessageByOutboxStatus(OutboxStatus outboxStatus) {
+        return orderOutboxRepository.findByTypeAndOutboxStatus(ORDER_PROCESSING_SAGA, outboxStatus);
+    }
+
+    @Transactional
+    public void deleteOrderOutboxMessageByOutboxStatus(OutboxStatus outboxStatus) {
+        orderOutboxRepository.deleteByTypeAndOutboxStatus(ORDER_PROCESSING_SAGA, outboxStatus);
+    }
+
+    @Transactional
+    public void saveOrderOutboxMessage(OrderEventPayload orderEventPayload,
+                                       PaymentStatus paymentStatus,
+                                       OutboxStatus outboxStatus,
+                                       UUID sagaId) {
+        save(OrderOutboxMessage.builder()
+                .id(UUID.randomUUID())
+                .sagaId(sagaId)
+                .createdAt(orderEventPayload.getCreatedAt())
+                .processedAt(ZonedDateTime.now(ZoneId.of(UTC)))
+                .type(ORDER_PROCESSING_SAGA)
+                .payload(createPayload(orderEventPayload))
+                .paymentStatus(paymentStatus)
+                .outboxStatus(outboxStatus)
+                .build());
+    }
+
+    @Transactional
+    public void updateOutboxMessage(OrderOutboxMessage orderOutboxMessage, OutboxStatus outboxStatus) {
+        orderOutboxMessage.setOutboxStatus(outboxStatus);
+        save(orderOutboxMessage);
+        log.info("Order outbox table status is updated as: {}", outboxStatus.name());
+    }
+
+    private String createPayload(OrderEventPayload orderEventPayload) {
+        try {
+            return objectMapper.writeValueAsString(orderEventPayload);
+        } catch (JsonProcessingException e) {
+            log.error("Could not create OrderEventPayload json!", e);
+            throw new PaymentDomainException("Could not create OrderEventPayload json!", e);
+        }
+    }
+
+    private void save(OrderOutboxMessage orderOutboxMessage) {
+        OrderOutboxMessage response = orderOutboxRepository.save(orderOutboxMessage);
+        if (Objects.isNull(response)) {
+            log.error("Could not save OrderOutboxMessage!");
+            throw new PaymentDomainException("Could not save OrderOutboxMessage!");
+        }
+        log.info("OrderOutboxMessage is saved with id: {}", orderOutboxMessage.getId());
+    }
+
+
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/outbox/scheduler/OrderOutboxScheduler.java

@@ -0,0 +1,44 @@
+package com.food.order.system.payment.application.service.outbox.scheduler;
+
+import com.food.order.system.outbox.OutboxScheduler;
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.payment.application.service.outbox.model.OrderOutboxMessage;
+import com.food.order.system.payment.application.service.ports.output.message.publisher.PaymentResponseMessagePublisher;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class OrderOutboxScheduler implements OutboxScheduler {
+
+    private final OrderOutboxHelper orderOutboxHelper;
+    private final PaymentResponseMessagePublisher paymentResponseMessagePublisher;
+
+
+    @Override
+    @Transactional
+    @Scheduled(fixedRateString = "${payment-service.outbox-scheduler-fixed-rate}",
+            initialDelayString = "${payment-service.outbox-scheduler-initial-delay}")
+    public void processOutboxMessage() {
+        var outboxMessagesResponse =
+                orderOutboxHelper.getOrderOutboxMessageByOutboxStatus(OutboxStatus.STARTED);
+        if (outboxMessagesResponse.isPresent() && outboxMessagesResponse.get().size() > 0) {
+            List<OrderOutboxMessage> outboxMessages = outboxMessagesResponse.get();
+            log.info("Received {} OrderOutboxMessage with ids {}, sending to message bus!", outboxMessages.size(),
+                    outboxMessages.stream().map(outboxMessage ->
+                            outboxMessage.getId().toString()).collect(Collectors.joining(",")));
+            outboxMessages.forEach(orderOutboxMessage ->
+                    paymentResponseMessagePublisher.publish(orderOutboxMessage,
+                            orderOutboxHelper::updateOutboxMessage));
+            log.info("{} OrderOutboxMessage sent to message bus!", outboxMessages.size());
+        }
+    }
+
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/ports/output/message/publisher/PaymentCancelledMessagePublisher.java

@@ -1,7 +0,0 @@
-package com.food.order.system.payment.application.service.ports.output.message.publisher;
-
-import com.food.order.system.payment.service.domain.event.PaymentCancelledEvent;
-import com.food.order.system.event.publisher.DomainEventPublisher;
-
-public interface PaymentCancelledMessagePublisher extends DomainEventPublisher<PaymentCancelledEvent> {
-}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/ports/output/message/publisher/PaymentCompletedMessagePublisher.java

@@ -1,7 +0,0 @@
-package com.food.order.system.payment.application.service.ports.output.message.publisher;
-
-import com.food.order.system.payment.service.domain.event.PaymentCompletedEvent;
-import com.food.order.system.event.publisher.DomainEventPublisher;
-
-public interface PaymentCompletedMessagePublisher extends DomainEventPublisher<PaymentCompletedEvent> {
-}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/ports/output/message/publisher/PaymentFailedMessagePublisher.java

@@ -1,7 +0,0 @@
-package com.food.order.system.payment.application.service.ports.output.message.publisher;
-
-import com.food.order.system.payment.service.domain.event.PaymentFailedEvent;
-import com.food.order.system.event.publisher.DomainEventPublisher;
-
-public interface PaymentFailedMessagePublisher extends DomainEventPublisher<PaymentFailedEvent> {
-}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/ports/output/message/publisher/PaymentResponseMessagePublisher.java

@@ -0,0 +1,11 @@
+package com.food.order.system.payment.application.service.ports.output.message.publisher;
+
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.payment.application.service.outbox.model.OrderOutboxMessage;
+
+import java.util.function.BiConsumer;
+
+public interface PaymentResponseMessagePublisher {
+    void publish(OrderOutboxMessage message,
+                 BiConsumer<OrderOutboxMessage , OutboxStatus> updateOutboxMessage);
+}

payment-service/payment-domain/payment-application-service/src/main/java/com/food/order/system/payment/application/service/ports/output/repository/OrderOutboxRepository.java

@@ -0,0 +1,26 @@
+package com.food.order.system.payment.application.service.ports.output.repository;
+
+import com.food.order.system.outbox.OutboxStatus;
+import com.food.order.system.payment.application.service.outbox.model.OrderOutboxMessage;
+import com.food.order.system.valueobject.PaymentStatus;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+
+@Repository
+public interface OrderOutboxRepository {
+
+    OrderOutboxMessage save(OrderOutboxMessage orderOutboxMessage);
+
+    Optional<List<OrderOutboxMessage>> findByTypeAndOutboxStatus(String type, OutboxStatus status);
+
+    Optional<OrderOutboxMessage> findByTypeAndSagaIdAndPaymentStatusAndOutboxStatus(String type,
+                                                                                    UUID sagaId,
+                                                                                    PaymentStatus paymentStatus,
+                                                                                    OutboxStatus outboxStatus);
+    void deleteByTypeAndOutboxStatus(String type, OutboxStatus status);
+
+
+}

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/PaymentDomainService.java

@@ -3,11 +3,7 @@ package com.food.order.system.payment.service.domain;
 import com.food.order.system.payment.service.domain.entity.CreditEntry;
 import com.food.order.system.payment.service.domain.entity.CreditHistory;
 import com.food.order.system.payment.service.domain.entity.Payment;
-import com.food.order.system.payment.service.domain.event.PaymentCancelledEvent;
-import com.food.order.system.payment.service.domain.event.PaymentCompletedEvent;
 import com.food.order.system.payment.service.domain.event.PaymentEvent;
-import com.food.order.system.payment.service.domain.event.PaymentFailedEvent;
-import com.food.order.system.event.publisher.DomainEventPublisher;
 
 import java.util.List;
 
@@ -16,14 +12,10 @@ public interface PaymentDomainService {
     PaymentEvent validateAndInitializePayment(  Payment paymentEvent,
                                                 CreditEntry creditEntry,
                                                 List<CreditHistory> creditHistory,
-                                                List<String> failureMessages,
-                                                DomainEventPublisher<PaymentCompletedEvent> publisher,
-                                                DomainEventPublisher<PaymentFailedEvent> failedEventDomainEventPublisher);
+                                                List<String> failureMessages);
 
     PaymentEvent validateAndCancelledPayment(  Payment paymentEvent,
                                                 CreditEntry creditEntry,
                                                 List<CreditHistory> creditHistory,
-                                                List<String> failureMessages,
-                                                DomainEventPublisher<PaymentCancelledEvent> publisher,
-                                                DomainEventPublisher<PaymentFailedEvent> failedEventDomainEventPublisher);
+                                                List<String> failureMessages);
 }

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/PaymentDomainServiceImpl.java

@@ -29,9 +29,7 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
     public PaymentEvent validateAndInitializePayment(Payment payment,
                                                      CreditEntry creditEntry,
                                                      List<CreditHistory> creditHistory,
-                                                     List<String> failureMessages,
-                                                     DomainEventPublisher<PaymentCompletedEvent> publisher,
-                                                     DomainEventPublisher<PaymentFailedEvent> failedPublisher) {
+                                                     List<String> failureMessages) {
         payment.validatePayment(failureMessages);
         payment.initializePayment();
         validateCreditEntry(payment,creditEntry,failureMessages);
@@ -42,11 +40,11 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
         if (failureMessages.isEmpty()) {
             log.info("Payment is valid and initialized");
             payment.updateStatus(PaymentStatus.COMPLETED);
-            return new PaymentCompletedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)),publisher );
+            return new PaymentCompletedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)) );
         } else {
             log.info("Payment is invalid and not initialized");
             payment.updateStatus(PaymentStatus.FAILED);
-            return new PaymentFailedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)), failureMessages,failedPublisher);
+            return new PaymentFailedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)), failureMessages);
         }
 
     }
@@ -55,9 +53,7 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
     public PaymentEvent validateAndCancelledPayment(Payment payment,
                                                     CreditEntry creditEntry,
                                                     List<CreditHistory> creditHistory,
-                                                    List<String> failureMessages,
-                                                    DomainEventPublisher<PaymentCancelledEvent> publisher,
-                                                    DomainEventPublisher<PaymentFailedEvent> failedPublisher) {
+                                                    List<String> failureMessages) {
 
         payment.validatePayment(failureMessages);
         addCreditEntry(payment,creditEntry);
@@ -66,11 +62,11 @@ public class PaymentDomainServiceImpl implements PaymentDomainService {
         if (failureMessages.isEmpty()) {
             log.info("Payment is valid and cancelled");
             payment.updateStatus(PaymentStatus.CANCELED);
-            return new PaymentCancelledEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)),publisher);
+            return new PaymentCancelledEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)));
         } else {
             log.info("Payment is invalid and not cancelled");
             payment.updateStatus(PaymentStatus.FAILED);
-            return new PaymentFailedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)), failureMessages,failedPublisher);
+            return new PaymentFailedEvent(payment, ZonedDateTime.now(ZoneId.of(UTC)), failureMessages);
         }
     }
 

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/event/PaymentCancelledEvent.java

@@ -1,23 +1,15 @@
 package com.food.order.system.payment.service.domain.event;
 
 import com.food.order.system.payment.service.domain.entity.Payment;
-import com.food.order.system.event.publisher.DomainEventPublisher;
 
 import java.time.ZonedDateTime;
 import java.util.Collections;
 
 public class PaymentCancelledEvent extends PaymentEvent{
 
-    private final DomainEventPublisher<PaymentCancelledEvent> publisher;
 
-    public PaymentCancelledEvent(Payment payment, ZonedDateTime createdAt, DomainEventPublisher<PaymentCancelledEvent> publisher) {
+    public PaymentCancelledEvent(Payment payment, ZonedDateTime createdAt) {
         super(payment, createdAt, Collections.emptyList());
-        this.publisher = publisher;
     }
 
-
-    @Override
-    public void fire() {
-        publisher.publish(this);
-    }
 }

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/event/PaymentCompletedEvent.java

@@ -8,18 +8,10 @@ import java.util.Collections;
 
 public class PaymentCompletedEvent extends PaymentEvent{
 
-    private final DomainEventPublisher<PaymentCompletedEvent> publisher;
 
     public PaymentCompletedEvent(Payment payment,
-                                 ZonedDateTime createdAt,
-                                 DomainEventPublisher<PaymentCompletedEvent> publisher) {
+                                 ZonedDateTime createdAt) {
         super(payment, createdAt , Collections.emptyList());
-        this.publisher = publisher;
     }
 
-
-    @Override
-    public void fire() {
-        publisher.publish(this);
-    }
 }

payment-service/payment-domain/payment-domain-core/src/main/java/com/food/order/system/payment/service/domain/event/PaymentFailedEvent.java

@@ -1,23 +1,15 @@
 package com.food.order.system.payment.service.domain.event;
 
 import com.food.order.system.payment.service.domain.entity.Payment;
-import com.food.order.system.event.publisher.DomainEventPublisher;
 
 import java.time.ZonedDateTime;
 import java.util.List;
 
 public class PaymentFailedEvent extends PaymentEvent{
 
-    private final DomainEventPublisher<PaymentFailedEvent> publisher;
 
-    public PaymentFailedEvent(Payment payment, ZonedDateTime createdAt, List<String> failureMessages,
-                              DomainEventPublisher<PaymentFailedEvent> publisher) {
+    public PaymentFailedEvent(Payment payment, ZonedDateTime createdAt, List<String> failureMessages) {
         super(payment, createdAt, failureMessages);
-        this.publisher = publisher;
     }
 
-    @Override
-    public void fire() {
-        publisher.publish(this);
-    }
 }

payment-service/payment-messaging/src/main/java/com/food/order/system/payment/messaging/publisher/kafka/PaymentCancelledKafkaMessagePublisher.java

@@ -1,10 +1,10 @@
 package com.food.order.system.payment.messaging.publisher.kafka;
 
+import com.food.order.system.event.publisher.DomainEventPublisher;
 import com.food.order.system.kafka.order.avro.model.PaymentResponseAvroModel;
 import com.food.order.system.kafka.producer.KafkaMessageHelper;
 import com.food.order.system.kafka.producer.service.KafkaProducer;
 import com.food.order.system.payment.application.service.config.PaymentServiceConfigData;
-import com.food.order.system.payment.application.service.ports.output.message.publisher.PaymentCancelledMessagePublisher;
 import com.food.order.system.payment.messaging.mapper.PaymentMessagingDataMapper;
 import com.food.order.system.payment.service.domain.event.PaymentCancelledEvent;
 import lombok.RequiredArgsConstructor;
@@ -14,7 +14,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 @RequiredArgsConstructor
-public class PaymentCancelledKafkaMessagePublisher implements PaymentCancelledMessagePublisher {
+public class PaymentCancelledKafkaMessagePublisher implements DomainEventPublisher<PaymentCancelledEvent> {
 
     private final PaymentMessagingDataMapper paymentDataMapper;
     private final KafkaProducer<String , PaymentResponseAvroModel> kafkaProducer;

payment-service/payment-messaging/src/main/java/com/food/order/system/payment/messaging/publisher/kafka/PaymentCompletedKafkaMessagePublisher.java

@@ -1,10 +1,10 @@
 package com.food.order.system.payment.messaging.publisher.kafka;
 
+import com.food.order.system.event.publisher.DomainEventPublisher;
 import com.food.order.system.kafka.order.avro.model.PaymentResponseAvroModel;
 import com.food.order.system.kafka.producer.KafkaMessageHelper;
 import com.food.order.system.kafka.producer.service.KafkaProducer;
 import com.food.order.system.payment.application.service.config.PaymentServiceConfigData;
-import com.food.order.system.payment.application.service.ports.output.message.publisher.PaymentCompletedMessagePublisher;
 import com.food.order.system.payment.messaging.mapper.PaymentMessagingDataMapper;
 import com.food.order.system.payment.service.domain.event.PaymentCompletedEvent;
 import lombok.RequiredArgsConstructor;
@@ -14,7 +14,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 @RequiredArgsConstructor
-public class PaymentCompletedKafkaMessagePublisher implements PaymentCompletedMessagePublisher {
+public class PaymentCompletedKafkaMessagePublisher implements DomainEventPublisher<PaymentCompletedEvent> {
 
     private final PaymentMessagingDataMapper paymentDataMapper;
     private final KafkaProducer<String , PaymentResponseAvroModel> kafkaProducer;

payment-service/payment-messaging/src/main/java/com/food/order/system/payment/messaging/publisher/kafka/PaymentFailedKafkaMessagePublisher.java

@@ -1,10 +1,10 @@
 package com.food.order.system.payment.messaging.publisher.kafka;
 
+import com.food.order.system.event.publisher.DomainEventPublisher;
 import com.food.order.system.kafka.order.avro.model.PaymentResponseAvroModel;
 import com.food.order.system.kafka.producer.KafkaMessageHelper;
 import com.food.order.system.kafka.producer.service.KafkaProducer;
 import com.food.order.system.payment.application.service.config.PaymentServiceConfigData;
-import com.food.order.system.payment.application.service.ports.output.message.publisher.PaymentFailedMessagePublisher;
 import com.food.order.system.payment.messaging.mapper.PaymentMessagingDataMapper;
 import com.food.order.system.payment.service.domain.event.PaymentFailedEvent;
 import lombok.RequiredArgsConstructor;
@@ -14,7 +14,7 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 @RequiredArgsConstructor
-public class PaymentFailedKafkaMessagePublisher implements PaymentFailedMessagePublisher {
+public class PaymentFailedKafkaMessagePublisher implements DomainEventPublisher<PaymentFailedEvent> {
 
     private final PaymentMessagingDataMapper paymentDataMapper;
     private final KafkaProducer<String , PaymentResponseAvroModel> kafkaProducer;

pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>spring-boot-starter-parent</artifactId>
         <groupId>org.springframework.boot</groupId>
-        <version>2.6.7</version>
+        <version>2.7.1</version>
         <relativePath/>
     </parent>
     <modelVersion>4.0.0</modelVersion>