Spring Boot Saga Pattern

Spring Boot Saga Pattern

Overview

Implements distributed transactions across microservices using the Saga Pattern. Replaces two-phase commit with a sequence of local transactions and compensating actions. Supports choreography (event-driven) and orchestration (centralized coordinator) approaches with Kafka, RabbitMQ, or Axon Framework.

When to Use

• Building distributed transactions across multiple microservices
• Replacing two-phase commit (2PC) with a more scalable solution
• Handling transaction rollback when a service fails
• Ensuring eventual consistency in microservices architecture
• Implementing compensating transactions for failed operations
• Coordinating complex business processes spanning multiple services

Trigger phrases: distributed transactions, saga pattern, compensating transactions, microservices transaction, eventual consistency, rollback across services, orchestration pattern, choreography pattern

Instructions

1. Design Transaction Flow

Map the sequence of operations and their compensating transactions:

Order → Payment → Inventory → Shipment
  ↓        ↓        ↓          ↓
Cancel  Refund   Release    Cancel

Validation: Verify every forward step has a corresponding compensation.

2. Choose Implementation Approach

Validation: Review team expertise and system complexity before choosing.

3. Implement Services with Local Transactions

Each service completes its local ACID transaction atomically:

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafka;

    @Transactional
    public Order createOrder(CreateOrderCommand cmd) {
        Order order = orderRepository.save(new Order(cmd.orderId(), cmd.items()));
        kafka.send("order.created", new OrderCreatedEvent(order.getId(), order.getItems()));
        return order;
    }
}

Validation: Test that local transaction commits before event is published.

4. Implement Compensating Transactions

Every forward operation requires an idempotent compensation:

@Service
@RequiredArgsConstructor
public class PaymentService {
    private final PaymentRepository paymentRepository;
    private final KafkaTemplate<String, Object> kafka;

    public void processPayment(PaymentRequest request) {
        Payment payment = paymentRepository.save(new Payment(request.orderId(), request.amount()));
        kafka.send("payment.processed", new PaymentProcessedEvent(payment.getId(), request.orderId()));
    }

    @Transactional
    public void refundPayment(String paymentId) {
        paymentRepository.findById(paymentId)
            .ifPresent(p -> {
                p.setStatus(REFUNDED);
                paymentRepository.save(p);
                kafka.send("payment.refunded", new PaymentRefundedEvent(paymentId));
            });
    }
}

Validation: Confirm compensation can execute safely multiple times (idempotency).

5. Set Up Message Broker

Configure Kafka with idempotent consumers:

@Configuration
@EnableKafka
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
            ConsumerFactory<String, Object> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(new DefaultErrorHandler());
        return factory;
    }
}

Validation: Enable transactional ID and verify exactly-once semantics.

6. Implement Saga Orchestrator (Orchestration Only)

@Service
@RequiredArgsConstructor
public class OrderSagaOrchestrator {
    private final KafkaTemplate<String, Object> kafka;
    private final SagaStateRepository sagaStateRepo;

    public void startSaga(OrderRequest request) {
        String sagaId = UUID.randomUUID().toString();
        sagaStateRepo.save(new SagaState(sagaId, STARTED, LocalDateTime.now()));
        kafka.send("saga.order.start", new StartOrderSagaCommand(sagaId, request));
    }

    @KafkaListener(topics = "payment.failed")
    public void handlePaymentFailed(PaymentFailedEvent event) {
        kafka.send("order.compensate", new CompensateOrderCommand(event.getSagaId()));
        kafka.send("inventory.compensate", new ReleaseInventoryCommand(event.getSagaId()));
        sagaStateRepo.updateStatus(event.getSagaId(), FAILED);
    }
}

Validation: Verify saga state persists before sending commands. Check compensation triggers on each failure path.

7. Implement Event Handlers (Choreography Only)

@Service
public class OrderEventHandler {
    private final OrderService orderService;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = "payment.processed", groupId = "order-service")
    public void onPaymentProcessed(PaymentProcessedEvent event) {
        try {
            InventoryReservedEvent result = orderService.reserveInventory(event.toInventoryRequest());
            kafka.send("inventory.reserved", result);
        } catch (InsufficientInventoryException e) {
            kafka.send("inventory.insufficient", new InsufficientInventoryEvent(event.getOrderId(), event.getPaymentId()));
        }
    }
}

Validation: Test that each event handler correctly triggers the next step or compensation.

8. Add Monitoring and Observability

@Configuration
public class SagaMetricsConfig {
    @Bean
    public MeterRegistry meterRegistry() {
        return new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
    }
}

Track: saga execution duration, compensation count, failure rate, stuck sagas.

Validation: Set up alerts for sagas exceeding expected duration.

Best Practices

Design:

• Make compensating transactions idempotent using database constraints or deduplication tables
• Use immutable events (Java records) to prevent accidental mutation
• Store saga state in persistent storage for recovery

Error Handling:

• Implement circuit breakers for inter-service calls
• Use dead-letter queues for messages exceeding retry limits
• Set appropriate timeouts per saga step (30s default, configurable)

Monitoring:

• Track saga status: PENDING, COMPLETED, COMPENSATING, FAILED
• Monitor compensation execution time
• Alert when sagas exceed SLA duration

Constraints and Warnings

• Every forward transaction MUST have a corresponding compensating transaction
• Compensating transactions MUST be idempotent to handle retry scenarios
• Saga state MUST be persisted to handle failures and recovery
• Never use synchronous communication between saga participants
• Sagas provide eventual consistency, not strong consistency
• Test all failure scenarios including partial failures
• Consider Axon Framework or Eventuate for complex orchestrations
• Ensure message brokers are highly available

Examples

Choreography-Based Saga

// Application.java
@SpringBootApplication
@EnableKafka
@EnableKafkaListeners
public class OrderApplication {
    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

// Event Classes (immutable)
public record OrderCreatedEvent(String orderId, List<OrderItem> items) {}
public record PaymentProcessedEvent(String paymentId, String orderId) {}
public record InventoryReservedEvent(String reservationId, String orderId) {}
public record PaymentFailedEvent(String orderId, String reason) {}
public record InsufficientInventoryEvent(String orderId, String paymentId) {}

// OrderService with compensation
@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = "payment.failed", groupId = "order-service")
    public void handleCompensation(PaymentFailedEvent event) {
        orderRepository.findByOrderId(event.orderId())
            .ifPresent(order -> {
                order.setStatus(CANCELLED);
                orderRepository.save(order);
            });
    }
}

Orchestration-Based Saga with Axon Framework

// Command
@Aggregate
public class OrderAggregate {
    @AggregateIdentifier
    private String orderId;

    @CommandHandler
    public OrderAggregate(CreateOrderCommand cmd) {
        apply(new OrderCreatedEvent(cmd.orderId(), cmd.items()));
    }

    @EventSourcingHandler
    public void on(OrderCreatedEvent event) {
        this.orderId = event.orderId();
    }

    @CommandHandler
    public void handle(CancelOrderCommand cmd) {
        apply(new OrderCancelledEvent(cmd.orderId(), cmd.reason()));
    }
}

References

• Saga Pattern Definition
• Choreography Implementation
• Orchestration Implementation
• Compensating Transactions
• State Management
• Error Handling and Retry
• Testing Strategies
• Pitfalls and Solutions
• Examples

Choreography-Based Saga Implementation

Architecture Overview

In choreography-based sagas, each service produces and listens to events. Services know what to do when they receive an event. No central coordinator manages the flow.

Service A → Event → Service B → Event → Service C
    ↓                   ↓                   ↓
  Event              Event               Event
    ↓                   ↓                   ↓
Compensation    Compensation        Compensation

Event Flow

Success Path

1. Order Service creates order → publishes OrderCreated event 2. Payment Service listens → processes payment → publishes PaymentProcessed event 3. Inventory Service listens → reserves inventory → publishes InventoryReserved event 4. Shipment Service listens → prepares shipment → publishes ShipmentPrepared event

Failure Path (When Payment Fails)

1. Payment Service publishes PaymentFailed event 2. Order Service listens → cancels order → publishes OrderCancelled event 3. All other services respond to cancellation with cleanup

Event Publisher

@Component
public class OrderEventPublisher {
    private final StreamBridge streamBridge;

    public OrderEventPublisher(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    public void publishOrderCreatedEvent(String orderId, BigDecimal amount, String itemId) {
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, amount, itemId);
        streamBridge.send("orderCreated-out-0",
            MessageBuilder
                .withPayload(event)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
    }
}

Event Listener

@Component
public class PaymentEventListener {

    @Bean
    public Consumer<OrderCreatedEvent> handleOrderCreatedEvent() {
        return event -> processPayment(event.getOrderId());
    }

    private void processPayment(String orderId) {
        // Payment processing logic
    }
}

Event Classes

public record OrderCreatedEvent(
    String orderId,
    BigDecimal amount,
    String itemId
) {}

public record PaymentProcessedEvent(
    String paymentId,
    String orderId,
    String itemId
) {}

public record PaymentFailedEvent(
    String paymentId,
    String orderId,
    String itemId,
    String reason
) {}

Spring Cloud Stream Configuration

spring:
  cloud:
    stream:
      bindings:
        orderCreated-out-0:
          destination: order-events
        paymentProcessed-out-0:
          destination: payment-events
        paymentFailed-out-0:
          destination: payment-events
      kafka:
        binder:
          brokers: localhost:9092

Maven Dependencies

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

Gradle Dependencies

implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'

Advantages and Disadvantages

Advantages

• Simple for small number of services
• Loose coupling between services
• No single point of failure
• Each service is independently deployable

Disadvantages

• Difficult to track workflow state - distributed across services
• Hard to troubleshoot - following event flow is complex
• Complexity grows with number of services
• Distributed source of truth - saga state not centralized

When to Use Choreography

Use choreography-based sagas when:

• Microservices are few in number (< 5 services per saga)
• Loose coupling is critical
• Team is experienced with event-driven architecture
• System can handle eventual consistency
• Workflow doesn't need centralized monitoring

Compensating Transactions

Design Principles

Idempotency

Execute multiple times with same result:

public void cancelPayment(String paymentId) {
    Payment payment = paymentRepository.findById(paymentId)
        .orElse(null);

    if (payment == null) {
        // Already cancelled or doesn't exist
        return;
    }

    if (payment.getStatus() == PaymentStatus.CANCELLED) {
        // Already cancelled, idempotent
        return;
    }

    payment.setStatus(PaymentStatus.CANCELLED);
    paymentRepository.save(payment);

    // Refund logic here
}

Retryability

Design operations to handle retries without side effects:

@Retryable(
    value = {TransientException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000, multiplier = 2)
)
public void releaseInventory(String itemId, int quantity) {
    // Use set operations for idempotency
    InventoryItem item = inventoryRepository.findById(itemId)
        .orElseThrow();

    item.increaseAvailableQuantity(quantity);
    inventoryRepository.save(item);
}

Compensation Strategies

Backward Recovery

Undo completed steps in reverse order:

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentFailedEvent event) {
    logger.error("Payment failed, initiating compensation");

    // Step 1: Cancel shipment preparation
    commandGateway.send(new CancelShipmentCommand(event.getOrderId()));

    // Step 2: Release inventory
    commandGateway.send(new ReleaseInventoryCommand(event.getOrderId()));

    // Step 3: Cancel order
    commandGateway.send(new CancelOrderCommand(event.getOrderId()));

    end();
}

Forward Recovery

Retry failed operation with exponential backoff:

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentTransientFailureEvent event) {
    if (event.getRetryCount() < MAX_RETRIES) {
        // Retry payment with backoff
        ProcessPaymentCommand retryCommand = new ProcessPaymentCommand(
            event.getPaymentId(),
            event.getOrderId(),
            event.getAmount()
        );
        commandGateway.send(retryCommand);
    } else {
        // After max retries, compensate
        handlePaymentFailure(event);
    }
}

Semantic Lock Pattern

Prevent concurrent modifications during saga execution:

@Entity
public class Order {
    @Id
    private String orderId;

    @Enumerated(EnumType.STRING)
    private OrderStatus status;

    @Version
    private Long version;

    private Instant lockedUntil;

    public boolean tryLock(Duration lockDuration) {
        if (isLocked()) {
            return false;
        }
        this.lockedUntil = Instant.now().plus(lockDuration);
        return true;
    }

    public boolean isLocked() {
        return lockedUntil != null &&
               Instant.now().isBefore(lockedUntil);
    }

    public void unlock() {
        this.lockedUntil = null;
    }
}

Compensation in Axon Framework

@Saga
public class OrderSaga {

    private String orderId;
    private String paymentId;
    private String inventoryId;
    private boolean compensating = false;

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(InventoryReservationFailedEvent event) {
        logger.error("Inventory reservation failed");
        compensating = true;

        // Compensate: refund payment
        RefundPaymentCommand refundCommand = new RefundPaymentCommand(
            paymentId,
            event.getOrderId(),
            event.getReservedAmount(),
            "Inventory unavailable"
        );

        commandGateway.send(refundCommand);
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentRefundedEvent event) {
        if (!compensating) return;

        logger.info("Payment refunded, cancelling order");

        // Compensate: cancel order
        CancelOrderCommand command = new CancelOrderCommand(
            event.getOrderId(),
            "Inventory unavailable - payment refunded"
        );

        commandGateway.send(command);
    }

    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCancelledEvent event) {
        logger.info("Saga completed with compensation");
    }
}

Handling Compensation Failures

Handle cases where compensation itself fails:

@Service
public class CompensationService {

    private final DeadLetterQueueService dlqService;

    public void handleCompensationFailure(String sagaId, String step, Exception cause) {
        logger.error("Compensation failed for saga {} at step {}", sagaId, step, cause);

        // Send to dead letter queue for manual intervention
        dlqService.send(new FailedCompensation(
            sagaId,
            step,
            cause.getMessage(),
            Instant.now()
        ));

        // Create alert for operations team
        alertingService.alert(
            "Compensation Failure",
            "Saga " + sagaId + " failed compensation at " + step
        );
    }
}

Testing Compensation

Verify that compensation produces expected results:

@Test
void shouldCompensateWhenPaymentFails() {
    String orderId = "order-123";
    String paymentId = "payment-456";

    // Arrange: execute payment
    Payment payment = new Payment(paymentId, orderId, BigDecimal.TEN);
    paymentRepository.save(payment);
    orderRepository.save(new Order(orderId, OrderStatus.PENDING));

    // Act: compensate
    paymentService.cancelPayment(paymentId);

    // Assert: verify idempotency
    paymentService.cancelPayment(paymentId);

    Payment result = paymentRepository.findById(paymentId).orElseThrow();
    assertThat(result.getStatus()).isEqualTo(PaymentStatus.CANCELLED);
}

Common Compensation Patterns

Inventory Release

@Service
public class InventoryService {

    public void releaseInventory(String orderId) {
        Order order = orderRepository.findById(orderId).orElseThrow();

        order.getItems().forEach(item -> {
            InventoryItem inventoryItem = inventoryRepository
                .findById(item.getProductId())
                .orElseThrow();

            inventoryItem.increaseAvailableQuantity(item.getQuantity());
            inventoryRepository.save(inventoryItem);
        });
    }
}

Payment Refund

@Service
public class PaymentService {

    public void refundPayment(String paymentId) {
        Payment payment = paymentRepository.findById(paymentId)
            .orElseThrow();

        if (payment.getStatus() == PaymentStatus.PROCESSED) {
            payment.setStatus(PaymentStatus.REFUNDED);
            paymentGateway.refund(payment.getTransactionId());
            paymentRepository.save(payment);
        }
    }
}

Order Cancellation

@Service
public class OrderService {

    public void cancelOrder(String orderId, String reason) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow();

        order.setStatus(OrderStatus.CANCELLED);
        order.setCancellationReason(reason);
        order.setCancelledAt(Instant.now());

        orderRepository.save(order);
    }
}

Error Handling and Retry Strategies

Retry Configuration

Use Spring Retry for automatic retry logic:

@Configuration
@EnableRetry
public class RetryConfig {

    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(2000L); // 2 second delay

        ExponentialBackOffPolicy exponentialBackOff = new ExponentialBackOffPolicy();
        exponentialBackOff.setInitialInterval(1000L);
        exponentialBackOff.setMultiplier(2.0);
        exponentialBackOff.setMaxInterval(10000L);

        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        retryTemplate.setBackOffPolicy(exponentialBackOff);
        retryTemplate.setRetryPolicy(retryPolicy);

        return retryTemplate;
    }
}

Retry with @Retryable

@Service
public class OrderService {

    @Retryable(
        value = {TransientException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 1000, multiplier = 2)
    )
    public void processOrder(String orderId) {
        // Order processing logic
    }

    @Recover
    public void recover(TransientException ex, String orderId) {
        logger.error("Order processing failed after retries: {}", orderId, ex);
        // Fallback logic
    }
}

Circuit Breaker with Resilience4j

Prevent cascading failures:

@Configuration
public class CircuitBreakerConfig {

    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)  // Open after 50% failures
            .waitDurationInOpenState(Duration.ofMillis(1000))
            .slidingWindowSize(2)      // Check last 2 calls
            .build();

        return CircuitBreakerRegistry.of(config);
    }
}

@Service
public class PaymentService {

    private final CircuitBreaker circuitBreaker;

    public PaymentService(CircuitBreakerRegistry registry) {
        this.circuitBreaker = registry.circuitBreaker("payment");
    }

    public PaymentResult processPayment(PaymentRequest request) {
        return circuitBreaker.executeSupplier(
            () -> callPaymentGateway(request)
        );
    }

    private PaymentResult callPaymentGateway(PaymentRequest request) {
        // Call external payment gateway
        return new PaymentResult(...);
    }
}

Dead Letter Queue

Handle failed messages:

@Configuration
public class DeadLetterQueueConfig {

    @Bean
    public NewTopic deadLetterTopic() {
        return new NewTopic("saga-dlq", 1, (short) 1);
    }
}

@Component
public class SagaErrorHandler implements ConsumerAwareErrorHandler {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public void handle(Exception thrownException,
                      List<ConsumerRecord<?, ?>> records,
                      Consumer<?, ?> consumer,
                      MessageListenerContainer container) {

        records.forEach(record -> {
            logger.error("Processing failed for message: {}", record.key());
            kafkaTemplate.send("saga-dlq", record.key(), record.value());
        });
    }
}

Timeout Handling

Define and enforce timeout policies:

@Service
public class TimeoutHandler {

    private final SagaStateRepository sagaStateRepository;
    private static final Duration STEP_TIMEOUT = Duration.ofSeconds(30);

    @Scheduled(fixedDelay = 5000)
    public void checkForTimeouts() {
        Instant timeoutThreshold = Instant.now().minus(STEP_TIMEOUT);

        List<SagaState> timedOutSagas = sagaStateRepository
            .findByStatusAndUpdatedAtBefore(SagaStatus.PROCESSING, timeoutThreshold);

        timedOutSagas.forEach(saga -> {
            logger.warn("Saga {} timed out at step {}",
                saga.getSagaId(), saga.getCurrentStep());
            compensateSaga(saga);
        });
    }

    private void compensateSaga(SagaState saga) {
        saga.setStatus(SagaStatus.COMPENSATING);
        sagaStateRepository.save(saga);
    }
}

Exponential Backoff

Prevent overwhelming downstream services:

@Service
public class BackoffService {

    public Duration calculateBackoff(int attemptNumber) {
        long baseDelay = 1000; // 1 second
        long delay = baseDelay * (long) Math.pow(2, attemptNumber - 1);
        long maxDelay = 30000; // 30 seconds

        return Duration.ofMillis(Math.min(delay, maxDelay));
    }

    @Retryable(
        value = {ServiceUnavailableException.class},
        maxAttempts = 5,
        backoff = @Backoff(
            delay = 1000,
            multiplier = 2.0,
            maxDelay = 30000
        )
    )
    public void callExternalService() {
        // External service call
    }
}

Idempotent Retry

Ensure retries don't cause duplicate processing:

@Service
public class IdempotentPaymentService {

    private final PaymentRepository paymentRepository;
    private final Map<String, PaymentResult> processedPayments = new ConcurrentHashMap<>();

    public PaymentResult processPayment(String paymentId, BigDecimal amount) {
        // Check if already processed
        if (processedPayments.containsKey(paymentId)) {
            return processedPayments.get(paymentId);
        }

        // Check database
        Optional<Payment> existing = paymentRepository.findById(paymentId);
        if (existing.isPresent()) {
            return new PaymentResult(existing.get());
        }

        // Process payment
        PaymentResult result = callPaymentGateway(paymentId, amount);

        // Cache and persist
        processedPayments.put(paymentId, result);
        paymentRepository.save(new Payment(paymentId, amount, result.getStatus()));

        return result;
    }
}

Global Exception Handler

Centralize error handling:

@RestControllerAdvice
public class GlobalExceptionHandler {

    @ExceptionHandler(SagaExecutionException.class)
    public ResponseEntity<ErrorResponse> handleSagaError(
            SagaExecutionException ex) {

        return ResponseEntity
            .status(HttpStatus.UNPROCESSABLE_ENTITY)
            .body(new ErrorResponse(
                "SAGA_EXECUTION_FAILED",
                ex.getMessage(),
                ex.getSagaId()
            ));
    }

    @ExceptionHandler(ServiceUnavailableException.class)
    public ResponseEntity<ErrorResponse> handleServiceUnavailable(
            ServiceUnavailableException ex) {

        return ResponseEntity
            .status(HttpStatus.SERVICE_UNAVAILABLE)
            .body(new ErrorResponse(
                "SERVICE_UNAVAILABLE",
                "Required service is temporarily unavailable"
            ));
    }

    @ExceptionHandler(TimeoutException.class)
    public ResponseEntity<ErrorResponse> handleTimeout(
            TimeoutException ex) {

        return ResponseEntity
            .status(HttpStatus.REQUEST_TIMEOUT)
            .body(new ErrorResponse(
                "REQUEST_TIMEOUT",
                "Request timed out after " + ex.getDuration()
            ));
    }
}

public record ErrorResponse(
    String code,
    String message,
    String details
) {
    public ErrorResponse(String code, String message) {
        this(code, message, null);
    }
}

Monitoring Error Rates

Track failure metrics:

@Component
public class SagaErrorMetrics {

    private final MeterRegistry meterRegistry;

    public SagaErrorMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
    }

    public void recordSagaFailure(String sagaType) {
        Counter.builder("saga.failure")
            .tag("type", sagaType)
            .register(meterRegistry)
            .increment();
    }

    public void recordRetry(String sagaType) {
        Counter.builder("saga.retry")
            .tag("type", sagaType)
            .register(meterRegistry)
            .increment();
    }

    public void recordTimeout(String sagaType) {
        Counter.builder("saga.timeout")
            .tag("type", sagaType)
            .register(meterRegistry)
            .increment();
    }
}

Event-Driven Architecture in Sagas

Event Types

Domain Events

Represent business facts that happened within a service:

public record OrderCreatedEvent(
    String orderId,
    Instant createdAt,
    BigDecimal amount
) implements DomainEvent {}

Integration Events

Communication between bounded contexts (microservices):

public record PaymentRequestedEvent(
    String orderId,
    String paymentId,
    BigDecimal amount
) implements IntegrationEvent {}

Command Events

Request for action by another service:

public record ProcessPaymentCommand(
    String paymentId,
    String orderId,
    BigDecimal amount
) {}

Event Versioning

Handle event schema evolution using versioning:

public record OrderCreatedEventV1(
    String orderId,
    BigDecimal amount
) {}

public record OrderCreatedEventV2(
    String orderId,
    BigDecimal amount,
    String customerId,
    Instant timestamp
) {}

// Event Upcaster
public class OrderEventUpcaster implements EventUpcaster {
    @Override
    public Stream<IntermediateEventRepresentation> upcast(
        Stream<IntermediateEventRepresentation> eventStream) {

        return eventStream.map(event -> {
            if (event.getType().getName().equals("OrderCreatedEventV1")) {
                return upcastV1ToV2(event);
            }
            return event;
        });
    }
}

Event Store

Store all events for audit trail and recovery:

@Entity
@Table(name = "saga_events")
public class SagaEvent {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(nullable = false)
    private String sagaId;

    @Column(nullable = false)
    private String eventType;

    @Column(columnDefinition = "TEXT")
    private String payload;

    @Column(nullable = false)
    private Instant timestamp;

    @Column(nullable = false)
    private Integer version;
}

Event Publishing Patterns

Outbox Pattern (Transactional)

Ensure atomic update of database and event publishing:

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxRepository outboxRepository;

    @Transactional
    public void createOrder(CreateOrderRequest request) {
        // 1. Create and save order
        Order order = new Order(...);
        orderRepository.save(order);

        // 2. Create outbox entry in same transaction
        OutboxEntry entry = new OutboxEntry(
            "OrderCreated",
            order.getId(),
            new OrderCreatedEvent(...)
        );
        outboxRepository.save(entry);
    }
}

@Component
public class OutboxPoller {

    @Scheduled(fixedDelay = 1000)
    public void pollAndPublish() {
        List<OutboxEntry> unpublished = outboxRepository.findUnpublished();

        unpublished.forEach(entry -> {
            eventPublisher.publish(entry.getEvent());
            outboxRepository.markAsPublished(entry.getId());
        });
    }
}

Direct Publishing Pattern

Publish events immediately after transaction:

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final EventPublisher eventPublisher;

    @Transactional
    public void createOrder(CreateOrderRequest request) {
        Order order = new Order(...);
        orderRepository.save(order);

        // Publish event after transaction commits
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronization() {
                @Override
                public void afterCommit() {
                    eventPublisher.publish(new OrderCreatedEvent(...));
                }
            }
        );
    }
}

Event Sourcing

Store all state changes as events instead of current state:

Benefits:

• Complete audit trail
• Time-travel debugging
• Natural fit for sagas
• Event replay for recovery

Implementation:

@Entity
public class Order {

    @Id
    private String orderId;

    @OneToMany(cascade = CascadeType.ALL, orphanRemoval = true)
    private List<DomainEvent> events = new ArrayList<>();

    public void createOrder(...) {
        apply(new OrderCreatedEvent(...));
    }

    protected void apply(DomainEvent event) {
        if (event instanceof OrderCreatedEvent e) {
            this.orderId = e.orderId();
            this.status = OrderStatus.PENDING;
        }
        events.add(event);
    }

    public List<DomainEvent> getUncommittedEvents() {
        return new ArrayList<>(events);
    }

    public void clearUncommittedEvents() {
        events.clear();
    }
}

Event Ordering and Consistency

Maintain Event Order

Use partitioning to maintain order within a saga:

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
}

@Service
public class EventPublisher {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publish(DomainEvent event) {
        // Use sagaId as key to maintain order
        kafkaTemplate.send("events", event.getSagaId(), event);
    }
}

Handle Out-of-Order Events

Use saga state to detect and handle out-of-order events:

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentProcessedEvent event) {
    if (saga.getStatus() != SagaStatus.AWAITING_PAYMENT) {
        // Out of order event, ignore or queue for retry
        logger.warn("Unexpected event in state: {}", saga.getStatus());
        return;
    }
    // Process event
}

Spring Boot SAGA Pattern - Examples

Table of Contents

1. E-Commerce Order Processing 2. Food Delivery Application 3. Travel Booking System 4. Banking Transfer System 5. Microservices Choreography Example 6. Microservices Orchestration Example

---

E-Commerce Order Processing

Complete example of an order processing system using orchestration-based saga with Axon Framework.

Architecture Overview

Order Service → Payment Service → Inventory Service → Shipment Service → Notification Service
     ↓               ↓                  ↓                   ↓                    ↓
  Compensation   Compensation      Compensation        Compensation         Compensation

Project Structure

e-commerce-saga/
├── order-service/
│   ├── domain/
│   │   ├── model/
│   │   │   └── Order.java
│   │   ├── event/
│   │   │   ├── OrderCreatedEvent.java
│   │   │   └── OrderCancelledEvent.java
│   │   └── command/
│   │       ├── CreateOrderCommand.java
│   │       └── CancelOrderCommand.java
│   └── saga/
│       └── OrderSaga.java
├── payment-service/
│   ├── domain/
│   │   ├── model/
│   │   │   └── Payment.java
│   │   ├── event/
│   │   │   ├── PaymentProcessedEvent.java
│   │   │   └── PaymentFailedEvent.java
│   │   └── command/
│   │       ├── ProcessPaymentCommand.java
│   │       └── RefundPaymentCommand.java
│   └── aggregate/
│       └── PaymentAggregate.java
├── inventory-service/
├── shipment-service/
└── notification-service/

Domain Models

Order Entity

@Entity
@Table(name = "orders")
public class Order {
    
    @Id
    private String orderId;
    
    @Column(nullable = false)
    private String customerId;
    
    @Column(nullable = false)
    private BigDecimal totalAmount;
    
    @Enumerated(EnumType.STRING)
    @Column(nullable = false)
    private OrderStatus status;
    
    @OneToMany(cascade = CascadeType.ALL, orphanRemoval = true)
    private List<OrderItem> items = new ArrayList<>();
    
    @Column(nullable = false)
    private Instant createdAt;
    
    private Instant completedAt;
    
    @Version
    private Long version;
    
    // Constructor
    public Order() {
    }
    
    public Order(String orderId, String customerId, BigDecimal totalAmount) {
        this.orderId = orderId;
        this.customerId = customerId;
        this.totalAmount = totalAmount;
        this.status = OrderStatus.PENDING;
        this.createdAt = Instant.now();
    }
    
    // Business methods
    public void markAsProcessing() {
        if (this.status != OrderStatus.PENDING) {
            throw new IllegalStateException("Order must be pending to mark as processing");
        }
        this.status = OrderStatus.PROCESSING;
    }
    
    public void markAsCompleted() {
        if (this.status != OrderStatus.PROCESSING) {
            throw new IllegalStateException("Order must be processing to complete");
        }
        this.status = OrderStatus.COMPLETED;
        this.completedAt = Instant.now();
    }
    
    public void cancel() {
        if (this.status == OrderStatus.COMPLETED) {
            throw new IllegalStateException("Cannot cancel completed order");
        }
        this.status = OrderStatus.CANCELLED;
    }
    
    // Getters
    public String getOrderId() { return orderId; }
    public String getCustomerId() { return customerId; }
    public BigDecimal getTotalAmount() { return totalAmount; }
    public OrderStatus getStatus() { return status; }
    public List<OrderItem> getItems() { return items; }
    public Instant getCreatedAt() { return createdAt; }
    public Instant getCompletedAt() { return completedAt; }
}

public enum OrderStatus {
    PENDING,
    PROCESSING,
    COMPLETED,
    CANCELLED,
    FAILED
}

Order Item

@Entity
@Table(name = "order_items")
public class OrderItem {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false)
    private String productId;
    
    @Column(nullable = false)
    private String productName;
    
    @Column(nullable = false)
    private Integer quantity;
    
    @Column(nullable = false)
    private BigDecimal unitPrice;
    
    @Column(nullable = false)
    private BigDecimal totalPrice;
    
    // Constructors
    public OrderItem() {
    }
    
    public OrderItem(String productId, String productName, 
                     Integer quantity, BigDecimal unitPrice) {
        this.productId = productId;
        this.productName = productName;
        this.quantity = quantity;
        this.unitPrice = unitPrice;
        this.totalPrice = unitPrice.multiply(BigDecimal.valueOf(quantity));
    }
    
    // Getters
    public Long getId() { return id; }
    public String getProductId() { return productId; }
    public String getProductName() { return productName; }
    public Integer getQuantity() { return quantity; }
    public BigDecimal getUnitPrice() { return unitPrice; }
    public BigDecimal getTotalPrice() { return totalPrice; }
}

Commands and Events

Order Commands

public record CreateOrderCommand(
    @TargetAggregateIdentifier String orderId,
    String customerId,
    List<OrderItemDTO> items,
    BigDecimal totalAmount
) {}

public record CancelOrderCommand(
    @TargetAggregateIdentifier String orderId,
    String reason
) {}

public record CompleteOrderCommand(
    @TargetAggregateIdentifier String orderId
) {}

Order Events

public record OrderCreatedEvent(
    String orderId,
    String customerId,
    List<OrderItemDTO> items,
    BigDecimal totalAmount,
    Instant timestamp
) {}

public record OrderCancelledEvent(
    String orderId,
    String reason,
    Instant timestamp
) {}

public record OrderCompletedEvent(
    String orderId,
    Instant timestamp
) {}

Payment Commands

public record ProcessPaymentCommand(
    @TargetAggregateIdentifier String paymentId,
    String orderId,
    String customerId,
    BigDecimal amount,
    PaymentMethod paymentMethod
) {}

public record RefundPaymentCommand(
    @TargetAggregateIdentifier String paymentId,
    String orderId,
    BigDecimal amount,
    String reason
) {}

Payment Events

public record PaymentProcessedEvent(
    String paymentId,
    String orderId,
    String customerId,
    BigDecimal amount,
    Instant timestamp
) {}

public record PaymentFailedEvent(
    String paymentId,
    String orderId,
    String reason,
    Instant timestamp
) {}

public record PaymentRefundedEvent(
    String paymentId,
    String orderId,
    BigDecimal amount,
    Instant timestamp
) {}

Order Saga Implementation

@Saga
public class OrderSaga {
    
    private static final Logger logger = LoggerFactory.getLogger(OrderSaga.class);
    
    @Autowired
    private transient CommandGateway commandGateway;
    
    private String orderId;
    private String paymentId;
    private String shipmentId;
    private boolean compensating = false;
    
    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCreatedEvent event) {
        this.orderId = event.orderId();
        logger.info("Order saga started for orderId: {}", orderId);
        
        // Generate payment ID
        this.paymentId = UUID.randomUUID().toString();
        
        // Send process payment command
        ProcessPaymentCommand command = new ProcessPaymentCommand(
            paymentId,
            event.orderId(),
            event.customerId(),
            event.totalAmount(),
            new PaymentMethod("CREDIT_CARD", Map.of())
        );
        
        commandGateway.send(command, (commandMessage, commandResultMessage) -> {
            if (commandResultMessage.isExceptional()) {
                logger.error("Payment command failed for orderId: {}", orderId);
                // Handle command failure
            }
        });
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentProcessedEvent event) {
        logger.info("Payment processed for orderId: {}", event.orderId());
        
        if (compensating) {
            logger.info("Saga is compensating, skipping inventory reservation");
            return;
        }
        
        // Send reserve inventory command
        ReserveInventoryCommand command = new ReserveInventoryCommand(
            event.orderId(),
            event.orderId() // Assuming items are tracked by orderId
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(InventoryReservedEvent event) {
        logger.info("Inventory reserved for orderId: {}", event.orderId());
        
        if (compensating) {
            logger.info("Saga is compensating, skipping shipment preparation");
            return;
        }
        
        // Generate shipment ID
        this.shipmentId = UUID.randomUUID().toString();
        
        // Send prepare shipment command
        PrepareShipmentCommand command = new PrepareShipmentCommand(
            shipmentId,
            event.orderId(),
            event.orderId() // Assuming shipment details tracked by orderId
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(ShipmentPreparedEvent event) {
        logger.info("Shipment prepared for orderId: {}", event.orderId());
        
        if (compensating) {
            logger.info("Saga is compensating, skipping notification");
            return;
        }
        
        // Send notification command
        SendNotificationCommand command = new SendNotificationCommand(
            event.orderId(),
            "ORDER_CONFIRMED",
            "Your order has been confirmed and is being prepared for shipment."
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(NotificationSentEvent event) {
        logger.info("Notification sent for orderId: {}", event.orderId());
        
        // Complete the order
        CompleteOrderCommand command = new CompleteOrderCommand(event.orderId());
        commandGateway.send(command);
    }
    
    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCompletedEvent event) {
        logger.info("Order saga completed for orderId: {}", event.orderId());
    }
    
    // Compensation handlers
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentFailedEvent event) {
        logger.error("Payment failed for orderId: {}, reason: {}", 
            event.orderId(), event.reason());
        
        compensating = true;
        
        // Cancel the order
        CancelOrderCommand command = new CancelOrderCommand(
            event.orderId(),
            "Payment failed: " + event.reason()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(InventoryReservationFailedEvent event) {
        logger.error("Inventory reservation failed for orderId: {}, reason: {}", 
            event.orderId(), event.reason());
        
        compensating = true;
        
        // Refund payment
        RefundPaymentCommand refundCommand = new RefundPaymentCommand(
            paymentId,
            event.orderId(),
            event.amount(),
            "Inventory unavailable"
        );
        
        commandGateway.send(refundCommand);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentRefundedEvent event) {
        logger.info("Payment refunded for orderId: {}", event.orderId());
        
        // Cancel the order
        CancelOrderCommand command = new CancelOrderCommand(
            event.orderId(),
            "Inventory unavailable - payment refunded"
        );
        
        commandGateway.send(command);
    }
    
    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCancelledEvent event) {
        logger.info("Order saga ended with cancellation for orderId: {}", 
            event.orderId());
    }
}

Payment Aggregate

@Aggregate
public class PaymentAggregate {
    
    @AggregateIdentifier
    private String paymentId;
    
    private String orderId;
    private BigDecimal amount;
    private PaymentStatus status;
    
    public PaymentAggregate() {
    }
    
    @CommandHandler
    public PaymentAggregate(ProcessPaymentCommand command) {
        // Validate payment
        if (command.amount().compareTo(BigDecimal.ZERO) <= 0) {
            apply(new PaymentFailedEvent(
                command.paymentId(),
                command.orderId(),
                "Invalid payment amount",
                Instant.now()
            ));
            return;
        }
        
        // Simulate payment gateway call
        boolean paymentSuccessful = processPaymentWithGateway(command);
        
        if (paymentSuccessful) {
            apply(new PaymentProcessedEvent(
                command.paymentId(),
                command.orderId(),
                command.customerId(),
                command.amount(),
                Instant.now()
            ));
        } else {
            apply(new PaymentFailedEvent(
                command.paymentId(),
                command.orderId(),
                "Payment gateway declined",
                Instant.now()
            ));
        }
    }
    
    @EventSourcingHandler
    public void on(PaymentProcessedEvent event) {
        this.paymentId = event.paymentId();
        this.orderId = event.orderId();
        this.amount = event.amount();
        this.status = PaymentStatus.PROCESSED;
    }
    
    @EventSourcingHandler
    public void on(PaymentFailedEvent event) {
        this.paymentId = event.paymentId();
        this.orderId = event.orderId();
        this.status = PaymentStatus.FAILED;
    }
    
    @CommandHandler
    public void handle(RefundPaymentCommand command) {
        if (this.status != PaymentStatus.PROCESSED) {
            throw new IllegalStateException("Can only refund processed payments");
        }
        
        apply(new PaymentRefundedEvent(
            command.paymentId(),
            command.orderId(),
            command.amount(),
            Instant.now()
        ));
    }
    
    @EventSourcingHandler
    public void on(PaymentRefundedEvent event) {
        this.status = PaymentStatus.REFUNDED;
    }
    
    private boolean processPaymentWithGateway(ProcessPaymentCommand command) {
        // Simulate payment gateway integration
        // In real implementation, call actual payment gateway API
        try {
            // Simulate network delay
            Thread.sleep(100);
            
            // 90% success rate for demonstration
            return Math.random() > 0.1;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

enum PaymentStatus {
    PENDING,
    PROCESSED,
    FAILED,
    REFUNDED
}

Inventory Aggregate

@Aggregate
public class InventoryAggregate {
    
    @AggregateIdentifier
    private String inventoryId;
    
    private String orderId;
    private Map<String, Integer> reservedItems = new HashMap<>();
    
    public InventoryAggregate() {
    }
    
    @CommandHandler
    public InventoryAggregate(ReserveInventoryCommand command) {
        // Check inventory availability
        boolean available = checkInventoryAvailability(command.items());
        
        if (available) {
            apply(new InventoryReservedEvent(
                UUID.randomUUID().toString(),
                command.orderId(),
                command.items(),
                Instant.now()
            ));
        } else {
            apply(new InventoryReservationFailedEvent(
                command.orderId(),
                "Insufficient inventory",
                BigDecimal.ZERO, // Would calculate from order
                Instant.now()
            ));
        }
    }
    
    @EventSourcingHandler
    public void on(InventoryReservedEvent event) {
        this.inventoryId = event.inventoryId();
        this.orderId = event.orderId();
        this.reservedItems = event.items();
    }
    
    @CommandHandler
    public void handle(ReleaseInventoryCommand command) {
        apply(new InventoryReleasedEvent(
            this.inventoryId,
            command.orderId(),
            this.reservedItems,
            Instant.now()
        ));
    }
    
    @EventSourcingHandler
    public void on(InventoryReleasedEvent event) {
        this.reservedItems.clear();
    }
    
    private boolean checkInventoryAvailability(Map<String, Integer> items) {
        // In real implementation, check actual inventory database
        // For demonstration, 95% availability
        return Math.random() > 0.05;
    }
}

Order Service Implementation

@Service
public class OrderService {
    
    private final CommandGateway commandGateway;
    private final OrderRepository orderRepository;
    
    public OrderService(CommandGateway commandGateway, 
                       OrderRepository orderRepository) {
        this.commandGateway = commandGateway;
        this.orderRepository = orderRepository;
    }
    
    public String createOrder(CreateOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
        
        // Create order entity
        Order order = new Order(
            orderId,
            request.customerId(),
            request.totalAmount()
        );
        
        // Add order items
        request.items().forEach(item -> {
            order.getItems().add(new OrderItem(
                item.productId(),
                item.productName(),
                item.quantity(),
                item.unitPrice()
            ));
        });
        
        // Save order
        orderRepository.save(order);
        
        // Send create order command to start saga
        CreateOrderCommand command = new CreateOrderCommand(
            orderId,
            request.customerId(),
            request.items(),
            request.totalAmount()
        );
        
        commandGateway.send(command);
        
        return orderId;
    }
    
    public OrderDTO getOrder(String orderId) {
        Order order = orderRepository.findById(orderId)
            .orElseThrow(() -> new OrderNotFoundException(orderId));
        
        return OrderDTO.fromEntity(order);
    }
}

REST Controller

@RestController
@RequestMapping("/api/orders")
public class OrderController {
    
    private final OrderService orderService;
    
    public OrderController(OrderService orderService) {
        this.orderService = orderService;
    }
    
    @PostMapping
    public ResponseEntity<OrderResponse> createOrder(
            @Valid @RequestBody CreateOrderRequest request) {
        
        String orderId = orderService.createOrder(request);
        
        return ResponseEntity
            .status(HttpStatus.CREATED)
            .body(new OrderResponse(orderId, "Order created successfully"));
    }
    
    @GetMapping("/{orderId}")
    public ResponseEntity<OrderDTO> getOrder(@PathVariable String orderId) {
        OrderDTO order = orderService.getOrder(orderId);
        return ResponseEntity.ok(order);
    }
}

Configuration

Application Properties

# Application
spring.application.name=order-service
server.port=8080

# Database
spring.datasource.url=jdbc:postgresql://localhost:5432/orderdb
spring.datasource.username=orderuser
spring.datasource.password=orderpass
spring.jpa.hibernate.ddl-auto=update
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect

# Axon Configuration
axon.axonserver.servers=localhost:8124
axon.serializer.general=jackson
axon.serializer.events=jackson
axon.serializer.messages=jackson

# Actuator
management.endpoints.web.exposure.include=health,metrics,info,prometheus
management.endpoint.health.show-details=always
management.metrics.export.prometheus.enabled=true

Maven Dependencies

<dependencies>
    <!-- Spring Boot -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-validation</artifactId>
    </dependency>
    
    <!-- Axon Framework -->
    <dependency>
        <groupId>org.axonframework</groupId>
        <artifactId>axon-spring-boot-starter</artifactId>
        <version>4.9.0</version>
    </dependency>
    
    <!-- Database -->
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    
    <!-- Monitoring -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    
    <dependency>
        <groupId>io.micrometer</groupId>
        <artifactId>micrometer-registry-prometheus</artifactId>
    </dependency>
    
    <!-- Testing -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    
    <dependency>
        <groupId>org.axonframework</groupId>
        <artifactId>axon-test</artifactId>
        <version>4.9.0</version>
        <scope>test</scope>
    </dependency>
</dependencies>

---

Food Delivery Application

Choreography-based saga example using Spring Cloud Stream and Kafka.

Architecture

Order Service
     ↓ (publishes OrderCreatedEvent)
Restaurant Service
     ↓ (publishes OrderAcceptedEvent / OrderRejectedEvent)
Payment Service
     ↓ (publishes PaymentSuccessEvent / PaymentFailedEvent)
Delivery Service
     ↓ (publishes DeliveryAssignedEvent / DeliveryFailedEvent)
Notification Service

Domain Events

// Order Events
public record OrderCreatedEvent(
    String orderId,
    String customerId,
    String restaurantId,
    List<FoodItem> items,
    BigDecimal totalAmount,
    DeliveryAddress deliveryAddress,
    Instant timestamp
) {}

public record OrderCancelledEvent(
    String orderId,
    String reason,
    Instant timestamp
) {}

// Restaurant Events
public record OrderAcceptedEvent(
    String orderId,
    String restaurantId,
    int estimatedPreparationTime,
    Instant timestamp
) {}

public record OrderRejectedEvent(
    String orderId,
    String restaurantId,
    String reason,
    Instant timestamp
) {}

// Payment Events
public record PaymentSuccessEvent(
    String orderId,
    String paymentId,
    BigDecimal amount,
    Instant timestamp
) {}

public record PaymentFailedEvent(
    String orderId,
    String paymentId,
    String reason,
    Instant timestamp
) {}

// Delivery Events
public record DeliveryAssignedEvent(
    String orderId,
    String deliveryPersonId,
    String deliveryPersonName,
    Instant estimatedDeliveryTime,
    Instant timestamp
) {}

public record DeliveryFailedEvent(
    String orderId,
    String reason,
    Instant timestamp
) {}

Order Service

@Service
public class FoodOrderService {
    
    private final StreamBridge streamBridge;
    private final OrderRepository orderRepository;
    
    public FoodOrderService(StreamBridge streamBridge,
                           OrderRepository orderRepository) {
        this.streamBridge = streamBridge;
        this.orderRepository = orderRepository;
    }
    
    public String createOrder(CreateFoodOrderRequest request) {
        String orderId = UUID.randomUUID().toString();
        
        // Create and save order
        FoodOrder order = new FoodOrder(
            orderId,
            request.customerId(),
            request.restaurantId(),
            request.items(),
            calculateTotal(request.items()),
            request.deliveryAddress()
        );
        
        orderRepository.save(order);
        
        // Publish order created event
        OrderCreatedEvent event = new OrderCreatedEvent(
            orderId,
            request.customerId(),
            request.restaurantId(),
            request.items(),
            order.getTotalAmount(),
            request.deliveryAddress(),
            Instant.now()
        );
        
        streamBridge.send("order-events", event);
        
        return orderId;
    }
    
    @Bean
    public Consumer<OrderRejectedEvent> handleOrderRejected() {
        return event -> {
            FoodOrder order = orderRepository.findById(event.orderId())
                .orElseThrow();
            
            order.cancel("Restaurant rejected: " + event.reason());
            orderRepository.save(order);
            
            // Publish cancellation event
            OrderCancelledEvent cancelEvent = new OrderCancelledEvent(
                event.orderId(),
                event.reason(),
                Instant.now()
            );
            
            streamBridge.send("order-events", cancelEvent);
        };
    }
    
    @Bean
    public Consumer<PaymentFailedEvent> handlePaymentFailed() {
        return event -> {
            FoodOrder order = orderRepository.findById(event.orderId())
                .orElseThrow();
            
            order.cancel("Payment failed: " + event.reason());
            orderRepository.save(order);
            
            // Notify restaurant to cancel preparation
            RestaurantCancelOrderEvent cancelEvent = 
                new RestaurantCancelOrderEvent(
                    event.orderId(),
                    "Payment failed",
                    Instant.now()
                );
            
            streamBridge.send("restaurant-events", cancelEvent);
        };
    }
    
    private BigDecimal calculateTotal(List<FoodItem> items) {
        return items.stream()
            .map(item -> item.price().multiply(BigDecimal.valueOf(item.quantity())))
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    }
}

Restaurant Service

@Service
public class RestaurantService {
    
    private final StreamBridge streamBridge;
    private final RestaurantRepository restaurantRepository;
    
    public RestaurantService(StreamBridge streamBridge,
                            RestaurantRepository restaurantRepository) {
        this.streamBridge = streamBridge;
        this.restaurantRepository = restaurantRepository;
    }
    
    @Bean
    public Consumer<OrderCreatedEvent> handleOrderCreated() {
        return event -> {
            Restaurant restaurant = restaurantRepository
                .findById(event.restaurantId())
                .orElseThrow();
            
            // Check if restaurant can accept order
            if (restaurant.canAcceptOrder(event.items())) {
                // Accept order
                int preparationTime = restaurant.estimatePreparationTime(event.items());
                
                OrderAcceptedEvent acceptedEvent = new OrderAcceptedEvent(
                    event.orderId(),
                    event.restaurantId(),
                    preparationTime,
                    Instant.now()
                );
                
                streamBridge.send("restaurant-events", acceptedEvent);
                
                // Start preparing order
                restaurant.startPreparingOrder(event.orderId(), event.items());
                restaurantRepository.save(restaurant);
                
            } else {
                // Reject order
                OrderRejectedEvent rejectedEvent = new OrderRejectedEvent(
                    event.orderId(),
                    event.restaurantId(),
                    "Items unavailable or restaurant closed",
                    Instant.now()
                );
                
                streamBridge.send("restaurant-events", rejectedEvent);
            }
        };
    }
    
    @Bean
    public Consumer<RestaurantCancelOrderEvent> handleCancelOrder() {
        return event -> {
            Restaurant restaurant = restaurantRepository
                .findByOrderId(event.orderId())
                .orElse(null);
            
            if (restaurant != null) {
                restaurant.cancelOrderPreparation(event.orderId());
                restaurantRepository.save(restaurant);
            }
        };
    }
}

Payment Service

@Service
public class FoodPaymentService {
    
    private final StreamBridge streamBridge;
    private final PaymentGateway paymentGateway;
    
    public FoodPaymentService(StreamBridge streamBridge,
                             PaymentGateway paymentGateway) {
        this.streamBridge = streamBridge;
        this.paymentGateway = paymentGateway;
    }
    
    @Bean
    public Consumer<OrderAcceptedEvent> handleOrderAccepted() {
        return event -> {
            String paymentId = UUID.randomUUID().toString();
            
            try {
                // Process payment
                PaymentResult result = paymentGateway.processPayment(
                    paymentId,
                    event.orderId(),
                    getOrderAmount(event.orderId())
                );
                
                if (result.isSuccess()) {
                    PaymentSuccessEvent successEvent = new PaymentSuccessEvent(
                        event.orderId(),
                        paymentId,
                        result.amount(),
                        Instant.now()
                    );
                    
                    streamBridge.send("payment-events", successEvent);
                } else {
                    PaymentFailedEvent failedEvent = new PaymentFailedEvent(
                        event.orderId(),
                        paymentId,
                        result.errorMessage(),
                        Instant.now()
                    );
                    
                    streamBridge.send("payment-events", failedEvent);
                }
            } catch (Exception e) {
                PaymentFailedEvent failedEvent = new PaymentFailedEvent(
                    event.orderId(),
                    paymentId,
                    "Payment processing error: " + e.getMessage(),
                    Instant.now()
                );
                
                streamBridge.send("payment-events", failedEvent);
            }
        };
    }
    
    private BigDecimal getOrderAmount(String orderId) {
        // Fetch order amount from order service or database
        return BigDecimal.valueOf(50.00); // Placeholder
    }
}

Delivery Service

@Service
public class DeliveryService {
    
    private final StreamBridge streamBridge;
    private final DeliveryPersonRepository deliveryPersonRepository;
    
    public DeliveryService(StreamBridge streamBridge,
                          DeliveryPersonRepository deliveryPersonRepository) {
        this.streamBridge = streamBridge;
        this.deliveryPersonRepository = deliveryPersonRepository;
    }
    
    @Bean
    public Consumer<PaymentSuccessEvent> handlePaymentSuccess() {
        return event -> {
            // Find available delivery person
            DeliveryPerson deliveryPerson = 
                deliveryPersonRepository.findAvailableNearby()
                    .orElse(null);
            
            if (deliveryPerson != null) {
                // Assign delivery
                deliveryPerson.assignOrder(event.orderId());
                deliveryPersonRepository.save(deliveryPerson);
                
                Instant estimatedDelivery = Instant.now()
                    .plus(30, ChronoUnit.MINUTES);
                
                DeliveryAssignedEvent assignedEvent = new DeliveryAssignedEvent(
                    event.orderId(),
                    deliveryPerson.getId(),
                    deliveryPerson.getName(),
                    estimatedDelivery,
                    Instant.now()
                );
                
                streamBridge.send("delivery-events", assignedEvent);
            } else {
                // No delivery person available
                DeliveryFailedEvent failedEvent = new DeliveryFailedEvent(
                    event.orderId(),
                    "No delivery person available",
                    Instant.now()
                );
                
                streamBridge.send("delivery-events", failedEvent);
            }
        };
    }
}

Notification Service

@Service
public class NotificationService {
    
    private final EmailService emailService;
    private final SmsService smsService;
    
    public NotificationService(EmailService emailService, 
                              SmsService smsService) {
        this.emailService = emailService;
        this.smsService = smsService;
    }
    
    @Bean
    public Consumer<OrderAcceptedEvent> handleOrderAccepted() {
        return event -> {
            String message = String.format(
                "Your order #%s has been accepted by the restaurant. " +
                "Estimated preparation time: %d minutes",
                event.orderId(),
                event.estimatedPreparationTime()
            );
            
            sendNotification(event.orderId(), message);
        };
    }
    
    @Bean
    public Consumer<PaymentSuccessEvent> handlePaymentSuccess() {
        return event -> {
            String message = String.format(
                "Payment of $%.2f for order #%s processed successfully",
                event.amount(),
                event.orderId()
            );
            
            sendNotification(event.orderId(), message);
        };
    }
    
    @Bean
    public Consumer<DeliveryAssignedEvent> handleDeliveryAssigned() {
        return event -> {
            String message = String.format(
                "Your order #%s has been assigned to delivery person %s. " +
                "Estimated delivery: %s",
                event.orderId(),
                event.deliveryPersonName(),
                event.estimatedDeliveryTime()
            );
            
            sendNotification(event.orderId(), message);
        };
    }
    
    @Bean
    public Consumer<OrderCancelledEvent> handleOrderCancelled() {
        return event -> {
            String message = String.format(
                "Your order #%s has been cancelled. Reason: %s",
                event.orderId(),
                event.reason()
            );
            
            sendNotification(event.orderId(), message);
        };
    }
    
    private void sendNotification(String orderId, String message) {
        // Get customer contact info from order
        // Send email and SMS
        emailService.sendEmail(orderId, message);
        smsService.sendSms(orderId, message);
    }
}

Spring Cloud Stream Configuration

spring:
  cloud:
    stream:
      bindings:
        # Order Service
        order-events:
          destination: food-order-events
          contentType: application/json
        
        # Restaurant Service
        restaurant-events:
          destination: food-restaurant-events
          contentType: application/json
        
        # Payment Service
        payment-events:
          destination: food-payment-events
          contentType: application/json
        
        # Delivery Service
        delivery-events:
          destination: food-delivery-events
          contentType: application/json
      
      kafka:
        binder:
          brokers: localhost:9092
          auto-create-topics: true
        
        bindings:
          order-events:
            producer:
              configuration:
                acks: all
                retries: 3
          
          restaurant-events:
            consumer:
              configuration:
                max-poll-records: 10

---

Travel Booking System

Complex orchestration example with multiple compensations.

Architecture

Flight Booking → Hotel Booking → Car Rental → Payment → Confirmation
     ↓                ↓              ↓           ↓            ↓
Cancel Flight   Cancel Hotel   Cancel Car   Refund      Cancel All

Travel Saga

@Saga
public class TravelBookingSaga {
    
    @Autowired
    private transient CommandGateway commandGateway;
    
    private String bookingId;
    private String flightReservationId;
    private String hotelReservationId;
    private String carRentalReservationId;
    private String paymentId;
    private boolean compensating = false;
    
    @StartSaga
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(TravelBookingStartedEvent event) {
        this.bookingId = event.bookingId();
        
        // Step 1: Book flight
        this.flightReservationId = UUID.randomUUID().toString();
        
        BookFlightCommand command = new BookFlightCommand(
            flightReservationId,
            event.bookingId(),
            event.flightDetails()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(FlightBookedEvent event) {
        if (compensating) return;
        
        // Step 2: Book hotel
        this.hotelReservationId = UUID.randomUUID().toString();
        
        BookHotelCommand command = new BookHotelCommand(
            hotelReservationId,
            event.bookingId(),
            event.hotelDetails()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(HotelBookedEvent event) {
        if (compensating) return;
        
        // Step 3: Rent car
        this.carRentalReservationId = UUID.randomUUID().toString();
        
        RentCarCommand command = new RentCarCommand(
            carRentalReservationId,
            event.bookingId(),
            event.carRentalDetails()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(CarRentedEvent event) {
        if (compensating) return;
        
        // Step 4: Process payment
        this.paymentId = UUID.randomUUID().toString();
        
        ProcessTravelPaymentCommand command = new ProcessTravelPaymentCommand(
            paymentId,
            event.bookingId(),
            calculateTotalAmount()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(TravelPaymentProcessedEvent event) {
        if (compensating) return;
        
        // Step 5: Confirm booking
        ConfirmTravelBookingCommand command = new ConfirmTravelBookingCommand(
            event.bookingId()
        );
        
        commandGateway.send(command);
    }
    
    @EndSaga
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(TravelBookingConfirmedEvent event) {
        // Saga completed successfully
    }
    
    // Compensation handlers
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(FlightBookingFailedEvent event) {
        compensating = true;
        
        CancelTravelBookingCommand command = new CancelTravelBookingCommand(
            event.bookingId(),
            "Flight booking failed: " + event.reason()
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(HotelBookingFailedEvent event) {
        compensating = true;
        
        // Cancel flight
        CancelFlightCommand cancelFlight = new CancelFlightCommand(
            flightReservationId,
            event.bookingId()
        );
        
        commandGateway.send(cancelFlight);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(FlightCancelledEvent event) {
        if (!compensating) return;
        
        // After flight cancelled, cancel entire booking
        CancelTravelBookingCommand command = new CancelTravelBookingCommand(
            event.bookingId(),
            "Hotel booking failed - flight cancelled"
        );
        
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(CarRentalFailedEvent event) {
        compensating = true;
        
        // Cancel hotel
        CancelHotelCommand cancelHotel = new CancelHotelCommand(
            hotelReservationId,
            event.bookingId()
        );
        
        commandGateway.send(cancelHotel);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(HotelCancelledEvent event) {
        if (!compensating) return;
        
        // Cancel flight
        CancelFlightCommand cancelFlight = new CancelFlightCommand(
            flightReservationId,
            event.bookingId()
        );
        
        commandGateway.send(cancelFlight);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(TravelPaymentFailedEvent event) {
        compensating = true;
        
        // Cancel car rental
        CancelCarRentalCommand cancelCar = new CancelCarRentalCommand(
            carRentalReservationId,
            event.bookingId()
        );
        
        commandGateway.send(cancelCar);
    }
    
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(CarRentalCancelledEvent event) {
        if (!compensating) return;
        
        // Cancel hotel
        CancelHotelCommand cancelHotel = new CancelHotelCommand(
            hotelReservationId,
            event.bookingId()
        );
        
        commandGateway.send(cancelHotel);
    }
    
    @EndSaga
    @SagaEventHandler(associationProperty = "bookingId")
    public void handle(TravelBookingCancelledEvent event) {
        // Saga ended with cancellation
    }
    
    private BigDecimal calculateTotalAmount() {
        // Calculate total from all bookings
        return BigDecimal.valueOf(1500.00); // Placeholder
    }
}

This examples file demonstrates practical implementations of the Saga Pattern in various scenarios. Each example shows:

1. Complete domain models 2. Commands and events 3. Saga coordination logic 4. Compensation transactions 5. Service implementations 6. Configuration and dependencies

The examples progress from simple to complex, showing both choreography and orchestration approaches with realistic business scenarios.

Orchestration-Based Saga Implementation

Architecture Overview

A central orchestrator (Saga Coordinator) manages the entire transaction flow, sending commands to services and handling responses.

         Saga Orchestrator
         /     |      \
    Service A  Service B  Service C

Orchestrator Responsibilities

1. Command Dispatch: Send commands to services 2. Response Handling: Process service responses 3. State Management: Track saga execution state 4. Compensation Coordination: Trigger compensating transactions on failure 5. Timeout Management: Handle service timeouts 6. Retry Logic: Manage retry attempts

Axon Framework Implementation

Saga Class

@Saga
public class OrderSaga {

    @Autowired
    private transient CommandGateway commandGateway;

    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCreatedEvent event) {
        String paymentId = UUID.randomUUID().toString();
        ProcessPaymentCommand command = new ProcessPaymentCommand(
            paymentId,
            event.getOrderId(),
            event.getAmount(),
            event.getItemId()
        );
        commandGateway.send(command);
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentProcessedEvent event) {
        ReserveInventoryCommand command = new ReserveInventoryCommand(
            event.getOrderId(),
            event.getItemId()
        );
        commandGateway.send(command);
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentFailedEvent event) {
        CancelOrderCommand command = new CancelOrderCommand(event.getOrderId());
        commandGateway.send(command);
        end();
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(InventoryReservedEvent event) {
        PrepareShipmentCommand command = new PrepareShipmentCommand(
            event.getOrderId(),
            event.getItemId()
        );
        commandGateway.send(command);
    }

    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCompletedEvent event) {
        // Saga completed successfully
    }
}

Aggregate for Order Service

@Aggregate
public class OrderAggregate {

    @AggregateIdentifier
    private String orderId;

    private OrderStatus status;

    public OrderAggregate() {
    }

    @CommandHandler
    public OrderAggregate(CreateOrderCommand command) {
        apply(new OrderCreatedEvent(
            command.getOrderId(),
            command.getAmount(),
            command.getItemId()
        ));
    }

    @EventSourcingHandler
    public void on(OrderCreatedEvent event) {
        this.orderId = event.getOrderId();
        this.status = OrderStatus.PENDING;
    }

    @CommandHandler
    public void handle(CancelOrderCommand command) {
        apply(new OrderCancelledEvent(command.getOrderId()));
    }

    @EventSourcingHandler
    public void on(OrderCancelledEvent event) {
        this.status = OrderStatus.CANCELLED;
    }
}

Aggregate for Payment Service

@Aggregate
public class PaymentAggregate {

    @AggregateIdentifier
    private String paymentId;

    public PaymentAggregate() {
    }

    @CommandHandler
    public PaymentAggregate(ProcessPaymentCommand command) {
        this.paymentId = command.getPaymentId();

        if (command.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
            apply(new PaymentFailedEvent(
                command.getPaymentId(),
                command.getOrderId(),
                command.getItemId(),
                "Payment amount must be greater than zero"
            ));
        } else {
            apply(new PaymentProcessedEvent(
                command.getPaymentId(),
                command.getOrderId(),
                command.getItemId()
            ));
        }
    }
}

Axon Configuration

axon:
  serializer:
    general: jackson
    events: jackson
    messages: jackson
  eventhandling:
    processors:
      order-processor:
        mode: tracking
        source: eventBus
  axonserver:
    enabled: false

Maven Dependencies for Axon

<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-spring-boot-starter</artifactId>
    <version>4.9.0</version> // Use latest stable version
</dependency>

Advantages and Disadvantages

Advantages

• Centralized visibility - easy to see workflow status
• Easier to troubleshoot - single place to analyze flow
• Clear transaction flow - orchestrator defines sequence
• Simplified error handling - centralized compensation logic
• Better for complex workflows - easier to manage many steps

Disadvantages

• Orchestrator becomes single point of failure - can be mitigated with clustering
• Additional infrastructure component - more complexity in deployment
• Potential tight coupling - if orchestrator knows too much about services

Eventuate Tram Sagas

Eventuate Tram is an alternative to Axon for orchestration-based sagas:

<dependency>
    <groupId>io.eventuate.tram.sagas</groupId>
    <artifactId>eventuate-tram-sagas-spring-starter</artifactId>
    <version>0.28.0</version> // Use latest stable version
</dependency>

Camunda for BPMN-Based Orchestration

Use Camunda when visual workflow design is beneficial:

Features:

• Visual workflow design
• BPMN 2.0 standard
• Human tasks support
• Complex workflow modeling

Use When:

• Business process modeling needed
• Visual workflow design preferred
• Human approval steps required
• Complex orchestration logic

When to Use Orchestration

Use orchestration-based sagas when:

• Building brownfield applications with existing microservices
• Handling complex workflows with many steps
• Centralized control and monitoring is critical
• Organization wants clear visibility into saga execution
• Need for human intervention in workflow

Common Pitfalls and Solutions

Pitfall 1: Lost Messages

Problem

Messages get lost due to broker failures, network issues, or consumer crashes before acknowledgment.

Solution

Use persistent messages with acknowledgments:

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.ACKS_CONFIG, "all");        // All replicas must acknowledge
    config.put(ProducerConfig.RETRIES_CONFIG, 3);         // Retry failed sends
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // Prevent duplicates
    return new DefaultKafkaProducerFactory<>(config);
}

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // Manual commit
    return new DefaultKafkaConsumerFactory<>(config);
}

Prevention Checklist

• ✓ Configure producer to wait for all replicas (acks=all)
• ✓ Enable idempotence to prevent duplicate messages
• ✓ Use manual commit for consumers
• ✓ Monitor message lag and broker health
• ✓ Use transactional outbox pattern

---

Pitfall 2: Duplicate Processing

Problem

Same message processed multiple times due to failed acknowledgments or retries, causing side effects.

Solution

Implement idempotency with deduplication:

@Service
public class DeduplicationService {

    private final DeduplicationRepository repository;

    public boolean isDuplicate(String messageId) {
        return repository.existsById(messageId);
    }

    public void recordProcessed(String messageId) {
        DeduplicatedMessage entry = new DeduplicatedMessage(
            messageId,
            Instant.now()
        );
        repository.save(entry);
    }
}

@Component
public class PaymentEventListener {

    private final DeduplicationService deduplicationService;
    private final PaymentService paymentService;

    @Bean
    public Consumer<PaymentEvent> handlePaymentEvent() {
        return event -> {
            String messageId = event.getMessageId();

            if (deduplicationService.isDuplicate(messageId)) {
                logger.info("Duplicate message ignored: {}", messageId);
                return;
            }

            paymentService.processPayment(event);
            deduplicationService.recordProcessed(messageId);
        };
    }
}

Prevention Checklist

• ✓ Add unique message ID to all events
• ✓ Implement deduplication cache/database
• ✓ Make all operations idempotent
• ✓ Use version control for entity updates
• ✓ Test with message replay

---

Pitfall 3: Saga State Inconsistency

Problem

Saga state in database doesn't match actual service states, leading to orphaned or stuck sagas.

Solution

Use event sourcing or state reconciliation:

@Service
public class SagaStateReconciler {

    private final SagaStateRepository stateRepository;
    private final OrderRepository orderRepository;
    private final PaymentRepository paymentRepository;

    @Scheduled(fixedDelay = 60000) // Run every minute
    public void reconcileSagaStates() {
        List<SagaState> processingSagas = stateRepository
            .findByStatus(SagaStatus.PROCESSING);

        processingSagas.forEach(saga -> {
            if (isActuallyCompleted(saga)) {
                logger.info("Reconciling saga {} - marking as completed", saga.getSagaId());
                saga.setStatus(SagaStatus.COMPLETED);
                saga.setCompletedAt(Instant.now());
                stateRepository.save(saga);
            }
        });
    }

    private boolean isActuallyCompleted(SagaState saga) {
        String orderId = saga.getSagaId();

        Order order = orderRepository.findById(orderId).orElse(null);
        if (order == null || order.getStatus() != OrderStatus.COMPLETED) {
            return false;
        }

        Payment payment = paymentRepository.findByOrderId(orderId).orElse(null);
        if (payment == null || payment.getStatus() != PaymentStatus.PROCESSED) {
            return false;
        }

        return true;
    }
}

Prevention Checklist

• ✓ Use event sourcing for complete audit trail
• ✓ Implement state reconciliation job
• ✓ Add health checks for saga coordinator
• ✓ Monitor saga state transitions
• ✓ Persist compensation steps

---

Pitfall 4: Orchestrator Single Point of Failure

Problem

Orchestration-based saga fails when orchestrator is down, blocking all sagas.

Solution

Implement clustering and failover:

@Configuration
public class SagaOrchestratorClusterConfig {

    @Bean
    public SagaStateRepository sagaStateRepository() {
        // Use shared database for cluster-wide state
        return new DatabaseSagaStateRepository();
    }

    @Bean
    @Primary
    public CommandGateway clusterAwareCommandGateway(
            CommandBus commandBus) {

        return new ClusterAwareCommandGateway(commandBus);
    }
}

@Component
public class OrchestratorHealthCheck extends AbstractHealthIndicator {

    private final SagaStateRepository repository;

    @Override
    protected void doHealthCheck(Health.Builder builder) {
        long stuckSagas = repository.countStuckSagas(Duration.ofMinutes(30));

        if (stuckSagas > 100) {
            builder.down()
                .withDetail("stuckSagas", stuckSagas)
                .withDetail("severity", "critical");
        } else if (stuckSagas > 10) {
            builder.degraded()
                .withDetail("stuckSagas", stuckSagas)
                .withDetail("severity", "warning");
        } else {
            builder.up()
                .withDetail("stuckSagas", stuckSagas);
        }
    }
}

Prevention Checklist

• ✓ Deploy orchestrator in cluster with shared state
• ✓ Use distributed coordination (ZooKeeper, Consul)
• ✓ Implement heartbeat monitoring
• ✓ Set up automatic failover
• ✓ Use circuit breakers for service calls

---

Pitfall 5: Non-Idempotent Compensations

Problem

Compensation logic fails on retry because it's not idempotent, leaving system in inconsistent state.

Solution

Design all compensations to be idempotent:

// Bad - Not idempotent
@Service
public class BadPaymentService {
    public void refundPayment(String paymentId) {
        Payment payment = paymentRepository.findById(paymentId).orElseThrow();
        payment.setStatus(PaymentStatus.REFUNDED);
        paymentRepository.save(payment);

        // If this fails partway, retry causes problems
        externalPaymentGateway.refund(payment.getTransactionId());
    }
}

// Good - Idempotent
@Service
public class GoodPaymentService {
    public void refundPayment(String paymentId) {
        Payment payment = paymentRepository.findById(paymentId)
            .orElse(null);

        if (payment == null) {
            // Already deleted or doesn't exist
            logger.info("Payment {} not found, skipping refund", paymentId);
            return;
        }

        if (payment.getStatus() == PaymentStatus.REFUNDED) {
            // Already refunded
            logger.info("Payment {} already refunded", paymentId);
            return;
        }

        try {
            externalPaymentGateway.refund(payment.getTransactionId());
            payment.setStatus(PaymentStatus.REFUNDED);
            paymentRepository.save(payment);
        } catch (Exception e) {
            logger.error("Refund failed, will retry", e);
            throw e;
        }
    }
}

Prevention Checklist

• ✓ Check current state before making changes
• ✓ Use status flags to track compensation completion
• ✓ Make database updates idempotent
• ✓ Test compensation with replays
• ✓ Document compensation logic

---

Pitfall 6: Missing Timeouts

Problem

Sagas hang indefinitely waiting for events that never arrive due to service failures.

Solution

Implement timeout mechanisms:

@Configuration
public class SagaTimeoutConfig {

    @Bean
    public SagaLifecycle sagaLifecycle(SagaStateRepository repository) {
        return new SagaLifecycle() {
            @Override
            public void onSagaFinished(Saga saga) {
                // Update saga state
            }
        };
    }
}

@Saga
public class OrderSaga {

    @Autowired
    private transient CommandGateway commandGateway;

    private String orderId;
    private String paymentId;
    private DeadlineManager deadlineManager;

    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCreatedEvent event) {
        this.orderId = event.orderId();

        // Schedule timeout for payment processing
        deadlineManager.scheduleDeadline(
            Duration.ofSeconds(30),
            "PaymentTimeout",
            orderId
        );

        commandGateway.send(new ProcessPaymentCommand(...));
    }

    @DeadlineHandler(deadlineName = "PaymentTimeout")
    public void handlePaymentTimeout() {
        logger.warn("Payment processing timed out for order {}", orderId);

        // Compensate
        commandGateway.send(new CancelOrderCommand(orderId));
        end();
    }

    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentProcessedEvent event) {
        // Cancel timeout
        deadlineManager.cancelDeadline("PaymentTimeout", orderId);
        // Continue saga...
    }
}

Prevention Checklist

• ✓ Set timeout for each saga step
• ✓ Use deadline manager to track timeouts
• ✓ Cancel timeouts when step completes
• ✓ Log timeout events
• ✓ Alert operations on repeated timeouts

---

Pitfall 7: Tight Coupling Between Services

Problem

Saga logic couples services tightly, making independent deployment impossible.

Solution

Use event-driven communication:

// Bad - Tight coupling
@Service
public class TightlyAgedOrderService {
    public void createOrder(OrderRequest request) {
        Order order = orderRepository.save(new Order(...));

        // Direct coupling to payment service
        paymentService.processPayment(order.getId(), request.getAmount());
    }
}

// Good - Event-driven
@Service
public class LooselyAgedOrderService {
    public void createOrder(OrderRequest request) {
        Order order = orderRepository.save(new Order(...));

        // Publish event - services listen independently
        eventPublisher.publish(new OrderCreatedEvent(
            order.getId(),
            request.getAmount()
        ));
    }
}

@Component
public class PaymentServiceListener {

    @Bean
    public Consumer<OrderCreatedEvent> handleOrderCreated() {
        return event -> {
            // Payment service can be deployed independently
            paymentService.processPayment(
                event.orderId(),
                event.amount()
            );
        };
    }
}

Prevention Checklist

• ✓ Use events for inter-service communication
• ✓ Avoid direct service-to-service calls
• ✓ Define clear contracts for events
• ✓ Version events for backward compatibility
• ✓ Deploy services independently

---

Pitfall 8: Inadequate Monitoring

Problem

Sagas fail silently or get stuck without visibility, making troubleshooting impossible.

Solution

Implement comprehensive monitoring:

@Component
public class SagaMonitoring {

    private final MeterRegistry meterRegistry;

    @Bean
    public MeterBinder sagaMetrics(SagaStateRepository repository) {
        return (registry) -> {
            Gauge.builder("saga.active", repository::countByStatus)
                .description("Number of active sagas")
                .register(registry);

            Gauge.builder("saga.stuck", () ->
                repository.countStuckSagas(Duration.ofMinutes(30)))
                .description("Number of stuck sagas")
                .register(registry);
        };
    }

    public void recordSagaStart(String sagaType) {
        Counter.builder("saga.started")
            .tag("type", sagaType)
            .register(meterRegistry)
            .increment();
    }

    public void recordSagaCompletion(String sagaType, long durationMs) {
        Timer.builder("saga.duration")
            .tag("type", sagaType)
            .publishPercentiles(0.5, 0.95, 0.99)
            .register(meterRegistry)
            .record(Duration.ofMillis(durationMs));
    }

    public void recordSagaFailure(String sagaType, String reason) {
        Counter.builder("saga.failed")
            .tag("type", sagaType)
            .tag("reason", reason)
            .register(meterRegistry)
            .increment();
    }
}

Prevention Checklist

• ✓ Track saga state transitions
• ✓ Monitor step execution times
• ✓ Alert on stuck sagas
• ✓ Log all failures with details
• ✓ Use distributed tracing (Sleuth, Zipkin)
• ✓ Create dashboards for visibility

Spring Boot SAGA Pattern - Reference Documentation

Table of Contents

1. Saga Pattern Overview 2. Choreography-Based Saga 3. Orchestration-Based Saga 4. Spring Boot Integration 5. Saga Frameworks 6. Event-Driven Architecture 7. Compensating Transactions 8. State Management 9. Error Handling and Retry 10. Testing Strategies

---

Saga Pattern Overview

Definition

A Saga is a sequence of local transactions where each transaction updates data within a single service. Each local transaction publishes an event or message that triggers the next local transaction in the saga. If a local transaction fails, the saga executes compensating transactions to undo the changes made by preceding transactions.

Key Characteristics

Distributed Transactions: Spans multiple microservices, each with its own database.

Local Transactions: Each service performs its own ACID transaction.

Event-Driven: Services communicate through events or commands.

Compensations: Rollback mechanism using compensating transactions.

Eventual Consistency: System reaches a consistent state over time.

Saga vs Two-Phase Commit (2PC)

ACID vs BASE

ACID (Traditional Databases):

• Atomicity: All or nothing
• Consistency: Valid state transitions
• Isolation: Concurrent transactions don't interfere
• Durability: Committed data persists

BASE (Saga Pattern):

• Basically Available: System is available most of the time
• Soft state: State may change over time
• Eventual consistency: System becomes consistent eventually

---

Choreography-Based Saga

Architecture

Each service produces and listens to events. Services know what to do when they receive an event.

Service A → Event → Service B → Event → Service C
    ↓                   ↓                   ↓
  Event              Event               Event
    ↓                   ↓                   ↓
Compensation    Compensation        Compensation

Event Flow

Success Flow: 1. Order Service creates order → publishes OrderCreated event 2. Payment Service listens → processes payment → publishes PaymentProcessed event 3. Inventory Service listens → reserves inventory → publishes InventoryReserved event 4. Shipment Service listens → prepares shipment → publishes ShipmentPrepared event

Failure Flow (Payment fails): 1. Payment Service publishes PaymentFailed event 2. Order Service listens → cancels order → publishes OrderCancelled event

Implementation Components

Event Publisher

@Component
public class OrderEventPublisher {
    private final StreamBridge streamBridge;
    
    public OrderEventPublisher(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }
    
    public void publishOrderCreatedEvent(String orderId, BigDecimal amount, String itemId) {
        OrderCreatedEvent event = new OrderCreatedEvent(orderId, amount, itemId);
        streamBridge.send("orderCreated-out-0", 
            MessageBuilder
                .withPayload(event)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
    }
}

Event Listener

@Component
public class PaymentEventListener {
    
    @Bean
    public Consumer<OrderCreatedEvent> handleOrderCreatedEvent() {
        return event -> processPayment(event.getOrderId());
    }
    
    private void processPayment(String orderId) {
        // Payment processing logic
    }
}

Event Classes

public record OrderCreatedEvent(
    String orderId,
    BigDecimal amount,
    String itemId
) {}

public record PaymentProcessedEvent(
    String paymentId,
    String orderId,
    String itemId
) {}

public record PaymentFailedEvent(
    String paymentId,
    String orderId,
    String itemId,
    String reason
) {}

Spring Cloud Stream Configuration

spring:
  cloud:
    stream:
      bindings:
        orderCreated-out-0:
          destination: order-events
        paymentProcessed-out-0:
          destination: payment-events
        paymentFailed-out-0:
          destination: payment-events
      kafka:
        binder:
          brokers: localhost:9092

Maven Dependencies

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

Gradle Dependencies

implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'

---

Orchestration-Based Saga

Architecture

A central Saga Orchestrator coordinates the entire transaction flow, sending commands to services and handling responses.

         Saga Orchestrator
         /     |      \
    Service A  Service B  Service C

Orchestrator Responsibilities

1. Command Dispatch: Sends commands to services 2. Response Handling: Processes service responses 3. State Management: Tracks saga execution state 4. Compensation Coordination: Triggers compensating transactions on failure 5. Timeout Management: Handles service timeouts 6. Retry Logic: Manages retry attempts

Axon Framework Implementation

Saga Class

@Saga
public class OrderSaga {
    
    @Autowired
    private transient CommandGateway commandGateway;
    
    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCreatedEvent event) {
        String paymentId = UUID.randomUUID().toString();
        ProcessPaymentCommand command = new ProcessPaymentCommand(
            paymentId, 
            event.getOrderId(), 
            event.getAmount(), 
            event.getItemId()
        );
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentProcessedEvent event) {
        ReserveInventoryCommand command = new ReserveInventoryCommand(
            event.getOrderId(), 
            event.getItemId()
        );
        commandGateway.send(command);
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(PaymentFailedEvent event) {
        CancelOrderCommand command = new CancelOrderCommand(event.getOrderId());
        commandGateway.send(command);
        end();
    }
    
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(InventoryReservedEvent event) {
        PrepareShipmentCommand command = new PrepareShipmentCommand(
            event.getOrderId(), 
            event.getItemId()
        );
        commandGateway.send(command);
    }
    
    @EndSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCompletedEvent event) {
        // Saga completed successfully
    }
}

Aggregate for Order Service

@Aggregate
public class OrderAggregate {
    
    @AggregateIdentifier
    private String orderId;
    
    private OrderStatus status;
    
    public OrderAggregate() {
    }
    
    @CommandHandler
    public OrderAggregate(CreateOrderCommand command) {
        apply(new OrderCreatedEvent(
            command.getOrderId(), 
            command.getAmount(), 
            command.getItemId()
        ));
    }
    
    @EventSourcingHandler
    public void on(OrderCreatedEvent event) {
        this.orderId = event.getOrderId();
        this.status = OrderStatus.PENDING;
    }
    
    @CommandHandler
    public void handle(CancelOrderCommand command) {
        apply(new OrderCancelledEvent(command.getOrderId()));
    }
    
    @EventSourcingHandler
    public void on(OrderCancelledEvent event) {
        this.status = OrderStatus.CANCELLED;
    }
}

Aggregate for Payment Service

@Aggregate
public class PaymentAggregate {
    
    @AggregateIdentifier
    private String paymentId;
    
    public PaymentAggregate() {
    }
    
    @CommandHandler
    public PaymentAggregate(ProcessPaymentCommand command) {
        this.paymentId = command.getPaymentId();
        
        if (command.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
            apply(new PaymentFailedEvent(
                command.getPaymentId(),
                command.getOrderId(),
                command.getItemId(),
                "Payment amount must be greater than zero"
            ));
        } else {
            apply(new PaymentProcessedEvent(
                command.getPaymentId(),
                command.getOrderId(),
                command.getItemId()
            ));
        }
    }
}

Axon Configuration

axon:
  serializer:
    general: jackson
    events: jackson
    messages: jackson
  eventhandling:
    processors:
      order-processor:
        mode: tracking
        source: eventBus
  axonserver:
    enabled: false

Maven Dependencies for Axon

<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-spring-boot-starter</artifactId>
    <version>4.9.0</version>
</dependency>

---

Spring Boot Integration

Application Configuration

@SpringBootApplication
@EnableScheduling
public class SagaApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(SagaApplication.class, args);
    }
}

Kafka Configuration

@Configuration
public class KafkaConfig {
    
    @Bean
    public NewTopic orderTopic() {
        return new NewTopic("order-events", 3, (short) 1);
    }
    
    @Bean
    public NewTopic paymentTopic() {
        return new NewTopic("payment-events", 3, (short) 1);
    }
    
    @Bean
    public NewTopic inventoryTopic() {
        return new NewTopic("inventory-events", 3, (short) 1);
    }
}

Properties Configuration

# Application
spring.application.name=saga-service

# Kafka
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=saga-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# Database
spring.datasource.url=jdbc:postgresql://localhost:5432/sagadb
spring.datasource.username=saga
spring.datasource.password=saga
spring.jpa.hibernate.ddl-auto=update
spring.jpa.properties.hibernate.dialect=org.hibernate.dialect.PostgreSQLDialect

# Actuator
management.endpoints.web.exposure.include=health,metrics,prometheus
management.endpoint.health.show-details=always

---

Saga Frameworks

Axon Framework

Type: Orchestration-based

Features:

• Event sourcing support
• CQRS pattern implementation
• Saga state management
• Automatic compensation
• Built-in retry mechanisms

Use When:

• Complex domain logic
• Event sourcing is beneficial
• CQRS pattern is needed
• Mature framework is required

Eventuate Tram Sagas

Type: Orchestration-based

Features:

• Database-per-service support
• Transactional messaging
• Saga orchestration DSL
• Multiple messaging platforms

Use When:

• Existing JPA-based services
• Transactional outbox pattern needed
• Multiple message brokers support required

Camunda

Type: BPMN-based orchestration

Features:

• Visual workflow design
• BPMN 2.0 standard
• Human tasks support
• Complex workflow modeling

Use When:

• Business process modeling needed
• Visual workflow design preferred
• Human approval steps required
• Complex orchestration logic

Apache Camel Saga EIP

Type: Enterprise Integration Pattern

Features:

• Saga EIP implementation
• Multiple protocol support
• Route-based compensation
• Integration with multiple systems

Use When:

• Enterprise integration scenarios
• Multiple protocol support needed
• Existing Camel infrastructure

---

Event-Driven Architecture

Event Types

Domain Events: Represent business facts that happened

public record OrderCreatedEvent(
    String orderId,
    Instant createdAt,
    BigDecimal amount
) implements DomainEvent {}

Integration Events: Communication between bounded contexts

public record PaymentRequestedEvent(
    String orderId,
    String paymentId,
    BigDecimal amount
) implements IntegrationEvent {}

Command Events: Request for action

public record ProcessPaymentCommand(
    String paymentId,
    String orderId,
    BigDecimal amount
) {}

Event Versioning

public record OrderCreatedEventV1(
    String orderId,
    BigDecimal amount
) {}

public record OrderCreatedEventV2(
    String orderId,
    BigDecimal amount,
    String customerId,
    Instant timestamp
) {}

// Event Upcaster
public class OrderEventUpcaster implements EventUpcaster {
    @Override
    public Stream<IntermediateEventRepresentation> upcast(
        Stream<IntermediateEventRepresentation> eventStream) {
        
        return eventStream.map(event -> {
            if (event.getType().getName().equals("OrderCreatedEventV1")) {
                return upcastV1ToV2(event);
            }
            return event;
        });
    }
}

Event Store

@Entity
@Table(name = "saga_events")
public class SagaEvent {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    @Column(nullable = false)
    private String sagaId;
    
    @Column(nullable = false)
    private String eventType;
    
    @Column(columnDefinition = "TEXT")
    private String payload;
    
    @Column(nullable = false)
    private Instant timestamp;
    
    @Column(nullable = false)
    private Integer version;
}

---

Compensating Transactions

Design Principles

Idempotency: Execute multiple times with same result

public void cancelPayment(String paymentId) {
    Payment payment = paymentRepository.findById(paymentId)
        .orElse(null);
    
    if (payment == null) {
        // Already cancelled or doesn't exist
        return;
    }
    
    if (payment.getStatus() == PaymentStatus.CANCELLED) {
        // Already cancelled, idempotent
        return;
    }
    
    payment.setStatus(PaymentStatus.CANCELLED);
    paymentRepository.save(payment);
    
    // Refund logic here
}

Retryability: Safe to retry on failure

@Retryable(
    value = {TransientException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000, multiplier = 2)
)
public void releaseInventory(String itemId, int quantity) {
    // Implementation
}

Compensation Strategies

Backward Recovery: Undo completed steps

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentFailedEvent event) {
    // Step 1: Cancel shipment preparation
    commandGateway.send(new CancelShipmentCommand(event.getOrderId()));
    
    // Step 2: Release inventory
    commandGateway.send(new ReleaseInventoryCommand(event.getOrderId()));
    
    // Step 3: Cancel order
    commandGateway.send(new CancelOrderCommand(event.getOrderId()));
    
    end();
}

Forward Recovery: Retry failed operation

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentTransientFailureEvent event) {
    if (event.getRetryCount() < MAX_RETRIES) {
        // Retry payment
        ProcessPaymentCommand retryCommand = new ProcessPaymentCommand(
            event.getPaymentId(),
            event.getOrderId(),
            event.getAmount()
        );
        commandGateway.send(retryCommand);
    } else {
        // Compensate
        handlePaymentFailure(event);
    }
}

Semantic Lock Pattern

Prevent concurrent modifications during saga execution:

@Entity
public class Order {
    @Id
    private String orderId;
    
    @Enumerated(EnumType.STRING)
    private OrderStatus status;
    
    @Version
    private Long version;
    
    private Instant lockedUntil;
    
    public boolean tryLock(Duration lockDuration) {
        if (isLocked()) {
            return false;
        }
        this.lockedUntil = Instant.now().plus(lockDuration);
        return true;
    }
    
    public boolean isLocked() {
        return lockedUntil != null && 
               Instant.now().isBefore(lockedUntil);
    }
    
    public void unlock() {
        this.lockedUntil = null;
    }
}

---

State Management

Saga State

@Entity
@Table(name = "saga_state")
public class SagaState {
    
    @Id
    private String sagaId;
    
    @Enumerated(EnumType.STRING)
    private SagaStatus status;
    
    @Column(columnDefinition = "TEXT")
    private String currentStep;
    
    @Column(columnDefinition = "TEXT")
    private String compensationSteps;
    
    private Instant startedAt;
    private Instant completedAt;
    
    @Version
    private Long version;
}

public enum SagaStatus {
    STARTED,
    PROCESSING,
    COMPENSATING,
    COMPLETED,
    FAILED,
    CANCELLED
}

State Machine with Spring Statemachine

@Configuration
@EnableStateMachine
public class SagaStateMachineConfig 
    extends StateMachineConfigurerAdapter<SagaStatus, SagaEvent> {
    
    @Override
    public void configure(
        StateMachineStateConfigurer<SagaStatus, SagaEvent> states) 
        throws Exception {
        
        states
            .withStates()
            .initial(SagaStatus.STARTED)
            .states(EnumSet.allOf(SagaStatus.class))
            .end(SagaStatus.COMPLETED)
            .end(SagaStatus.FAILED);
    }
    
    @Override
    public void configure(
        StateMachineTransitionConfigurer<SagaStatus, SagaEvent> transitions) 
        throws Exception {
        
        transitions
            .withExternal()
                .source(SagaStatus.STARTED)
                .target(SagaStatus.PROCESSING)
                .event(SagaEvent.ORDER_CREATED)
            .and()
            .withExternal()
                .source(SagaStatus.PROCESSING)
                .target(SagaStatus.COMPLETED)
                .event(SagaEvent.ALL_STEPS_COMPLETED)
            .and()
            .withExternal()
                .source(SagaStatus.PROCESSING)
                .target(SagaStatus.COMPENSATING)
                .event(SagaEvent.STEP_FAILED)
            .and()
            .withExternal()
                .source(SagaStatus.COMPENSATING)
                .target(SagaStatus.FAILED)
                .event(SagaEvent.COMPENSATION_COMPLETED);
    }
}

---

Error Handling and Retry

Retry Configuration

@Configuration
@EnableRetry
public class RetryConfig {
    
    @Bean
    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();
        
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(2000L);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        return retryTemplate;
    }
}

Circuit Breaker with Resilience4j

@Configuration
public class CircuitBreakerConfig {
    
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            .failureRateThreshold(50)
            .waitDurationInOpenState(Duration.ofMillis(1000))
            .slidingWindowSize(2)
            .build();
        
        return CircuitBreakerRegistry.of(config);
    }
}

@Service
public class PaymentService {
    
    private final CircuitBreaker circuitBreaker;
    
    public PaymentService(CircuitBreakerRegistry registry) {
        this.circuitBreaker = registry.circuitBreaker("payment");
    }
    
    public PaymentResult processPayment(PaymentRequest request) {
        return circuitBreaker.executeSupplier(
            () -> callPaymentGateway(request)
        );
    }
}

Dead Letter Queue

@Configuration
public class DeadLetterQueueConfig {
    
    @Bean
    public NewTopic deadLetterTopic() {
        return new NewTopic("saga-dlq", 1, (short) 1);
    }
}

@Component
public class SagaErrorHandler implements ConsumerAwareErrorHandler {
    
    private final KafkaTemplate<String, Object> kafkaTemplate;
    
    @Override
    public void handle(Exception thrownException, 
                      List<ConsumerRecord<?, ?>> records,
                      Consumer<?, ?> consumer, 
                      MessageListenerContainer container) {
        
        records.forEach(record -> {
            kafkaTemplate.send("saga-dlq", record.key(), record.value());
        });
    }
}

---

Testing Strategies

Unit Testing Saga

@Test
void shouldCompensateWhenPaymentFails() {
    // Given
    OrderSaga saga = new OrderSaga();
    FixtureConfiguration<OrderSaga> fixture = new SagaTestFixture<>(OrderSaga.class);
    
    String orderId = UUID.randomUUID().toString();
    String paymentId = UUID.randomUUID().toString();
    
    // When
    fixture
        .givenNoPriorActivity()
        .whenPublishingA(new OrderCreatedEvent(orderId, BigDecimal.TEN, "item-1"))
        .expectDispatchedCommands(new ProcessPaymentCommand(paymentId, orderId, BigDecimal.TEN));
    
    // Then - payment fails
    fixture
        .whenPublishingA(new PaymentFailedEvent(paymentId, orderId, "item-1", "Insufficient funds"))
        .expectDispatchedCommands(new CancelOrderCommand(orderId));
}

Integration Testing with Testcontainers

@SpringBootTest
@Testcontainers
class SagaIntegrationTest {
    
    @Container
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("confluentinc/cp-kafka:7.4.0")
    );
    
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
        "postgres:15-alpine"
    );
    
    @DynamicPropertySource
    static void overrideProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("spring.datasource.username", postgres::getUsername);
        registry.add("spring.datasource.password", postgres::getPassword);
    }
    
    @Test
    void shouldCompleteOrderSagaSuccessfully() {
        // Test implementation
    }
}

Testing Idempotency

@Test
void compensationShouldBeIdempotent() {
    String paymentId = "payment-123";
    
    // Execute compensation first time
    paymentService.cancelPayment(paymentId);
    Payment firstResult = paymentRepository.findById(paymentId).orElseThrow();
    
    // Execute compensation second time
    paymentService.cancelPayment(paymentId);
    Payment secondResult = paymentRepository.findById(paymentId).orElseThrow();
    
    // Should produce same result
    assertThat(firstResult).isEqualTo(secondResult);
    assertThat(secondResult.getStatus()).isEqualTo(PaymentStatus.CANCELLED);
}

---

Monitoring and Observability

Micrometer Metrics

@Component
public class SagaMetrics {
    
    private final Counter sagaStarted;
    private final Counter sagaCompleted;
    private final Counter sagaFailed;
    private final Timer sagaDuration;
    
    public SagaMetrics(MeterRegistry registry) {
        this.sagaStarted = Counter.builder("saga.started")
            .description("Number of sagas started")
            .register(registry);
            
        this.sagaCompleted = Counter.builder("saga.completed")
            .description("Number of sagas completed successfully")
            .register(registry);
            
        this.sagaFailed = Counter.builder("saga.failed")
            .description("Number of sagas failed")
            .register(registry);
            
        this.sagaDuration = Timer.builder("saga.duration")
            .description("Saga execution duration")
            .register(registry);
    }
    
    public void recordSagaStart() {
        sagaStarted.increment();
    }
    
    public void recordSagaCompletion(Duration duration) {
        sagaCompleted.increment();
        sagaDuration.record(duration);
    }
    
    public void recordSagaFailure() {
        sagaFailed.increment();
    }
}

Distributed Tracing

@Configuration
public class TracingConfig {
    
    @Bean
    public Tracer tracer() {
        return new Tracer.Builder()
            .spanReporter(new ZipkinSpanReporter())
            .build();
    }
}

@Service
public class OrderService {
    
    @Autowired
    private Tracer tracer;
    
    public void createOrder(OrderRequest request) {
        Span span = tracer.newTrace().name("create-order").start();
        try (Tracer.SpanInScope ws = tracer.withSpanInScope(span)) {
            // Order creation logic
            span.tag("orderId", request.getOrderId());
        } finally {
            span.finish();
        }
    }
}

Health Checks

@Component
public class SagaHealthIndicator implements HealthIndicator {
    
    private final SagaStateRepository sagaStateRepository;
    
    @Override
    public Health health() {
        long stuckSagas = sagaStateRepository.countStuckSagas(
            Duration.ofMinutes(30)
        );
        
        if (stuckSagas > 10) {
            return Health.down()
                .withDetail("stuckSagas", stuckSagas)
                .build();
        }
        
        return Health.up()
            .withDetail("stuckSagas", stuckSagas)
            .build();
    }
}

---

Performance Considerations

Batch Processing

@Service
public class BatchSagaProcessor {
    
    @Scheduled(fixedDelay = 5000)
    public void processPendingSagas() {
        List<SagaState> pendingSagas = sagaStateRepository
            .findByStatus(SagaStatus.PROCESSING, PageRequest.of(0, 100));
        
        pendingSagas.forEach(this::processSaga);
    }
}

Parallel Execution

@SagaEventHandler(associationProperty = "orderId")
public void handle(PaymentProcessedEvent event) {
    // Execute inventory and notification in parallel
    CompletableFuture.allOf(
        CompletableFuture.runAsync(() -> 
            commandGateway.send(new ReserveInventoryCommand(event.getOrderId()))
        ),
        CompletableFuture.runAsync(() -> 
            commandGateway.send(new SendNotificationCommand(event.getOrderId()))
        )
    ).join();
}

Database Optimization

-- Index for saga state queries
CREATE INDEX idx_saga_state_status ON saga_state(status);
CREATE INDEX idx_saga_state_started_at ON saga_state(started_at);

-- Index for event store queries
CREATE INDEX idx_saga_events_saga_id ON saga_events(saga_id);
CREATE INDEX idx_saga_events_timestamp ON saga_events(timestamp);

---

Security Best Practices

Message Authentication

@Configuration
public class MessageSecurityConfig {
    
    @Bean
    public MessageSigningInterceptor messageSigningInterceptor() {
        return new MessageSigningInterceptor(secretKey);
    }
}

public class MessageSigningInterceptor implements ProducerInterceptor<String, Object> {
    
    @Override
    public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
        String signature = computeSignature(record.value());
        Headers headers = record.headers();
        headers.add("signature", signature.getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

Audit Logging

@Aspect
@Component
public class SagaAuditAspect {
    
    @Around("@annotation(SagaOperation)")
    public Object auditSagaOperation(ProceedingJoinPoint joinPoint) throws Throwable {
        String sagaId = extractSagaId(joinPoint);
        String operation = joinPoint.getSignature().getName();
        
        auditLog.info("Saga operation started: sagaId={}, operation={}", 
            sagaId, operation);
        
        try {
            Object result = joinPoint.proceed();
            auditLog.info("Saga operation completed: sagaId={}, operation={}", 
                sagaId, operation);
            return result;
        } catch (Exception e) {
            auditLog.error("Saga operation failed: sagaId={}, operation={}, error={}", 
                sagaId, operation, e.getMessage());
            throw e;
        }
    }
}

---

Common Pitfalls and Solutions

Pitfall 1: Lost Messages

Problem: Messages get lost due to broker failures.

Solution: Use persistent messages and acknowledgments.

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.ACKS_CONFIG, "all");
    config.put(ProducerConfig.RETRIES_CONFIG, 3);
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    return new DefaultKafkaProducerFactory<>(config);
}

Pitfall 2: Duplicate Processing

Problem: Same message processed multiple times.

Solution: Implement idempotency with deduplication.

@Service
public class DeduplicationService {
    
    private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();
    
    public boolean isDuplicate(String messageId) {
        return !processedMessageIds.add(messageId);
    }
}

Pitfall 3: Saga State Inconsistency

Problem: Saga state doesn't match actual service states.

Solution: Use event sourcing or state reconciliation.

@Scheduled(fixedDelay = 60000)
public void reconcileSagaStates() {
    List<SagaState> processingSagas = 
        sagaStateRepository.findByStatus(SagaStatus.PROCESSING);
    
    processingSagas.forEach(saga -> {
        if (isActuallyCompleted(saga)) {
            saga.setStatus(SagaStatus.COMPLETED);
            sagaStateRepository.save(saga);
        }
    });
}

---

Additional Resources

• Microservices.io - Saga Pattern
• Axon Framework Documentation
• Spring Cloud Stream Reference
• Eventuate Tram Documentation
• Camunda Platform
• Apache Camel Saga EIP