Saga Choreography with Kafka

// === ORDER SERVICE ===
@Service
public class OrderService {

    @Transactional
    public void createOrder(CreateOrderRequest req) {
        Order order = orderRepo.save(Order.pending(req));
        // Starts the saga: the payment service listens to this
        kafkaTemplate.send("orders.order.created",
            order.getId(), new OrderCreatedEvent(order));
    }

    // Compensating action if payment fails
    @KafkaListener(topics = "payments.payment.failed")
    @Transactional
    public void onPaymentFailed(PaymentFailedEvent event) {
        Order order = orderRepo.findById(event.getOrderId()).orElseThrow();
        order.cancel("Payment failed: " + event.getReason());
        orderRepo.save(order);
        kafkaTemplate.send("orders.order.cancelled", order.getId(), new OrderCancelledEvent(order));
    }

    @KafkaListener(topics = "payments.payment.completed")
    @Transactional
    public void onPaymentCompleted(PaymentCompletedEvent event) {
        Order order = orderRepo.findById(event.getOrderId()).orElseThrow();
        order.confirmPayment();
        orderRepo.save(order);
        // Next saga step: inventory reservation
        kafkaTemplate.send("inventory.reservation.requested",
            order.getId(), new ReservationRequest(order));
    }
}

// === PAYMENT SERVICE ===
@Service
public class PaymentService {

    @KafkaListener(topics = "orders.order.created")
    public void onOrderCreated(OrderCreatedEvent event) {
        try {
            PaymentResult result = paymentGateway.charge(event.getAmount(), event.getCustomerId());
            kafkaTemplate.send("payments.payment.completed",
                event.getOrderId(), new PaymentCompletedEvent(event, result));
        } catch (PaymentException e) {
            kafkaTemplate.send("payments.payment.failed",
                event.getOrderId(), new PaymentFailedEvent(event, e.getMessage()));
        }
    }
}