// Add a saga_in_progress flag to the order
public class Order {
    private OrderStatus status;
    private boolean sagaInProgress;  // true while saga is active
}
 
// Downstream systems check this flag
public OrderSummary getOrderSummary(String orderId) {
    Order order = orderRepository.findById(orderId).orElseThrow();
 
    if (order.isSagaInProgress()) {
        // Return a safe "processing" representation
        return OrderSummary.processing(orderId);
    }
 
    return OrderSummary.from(order);
}

// For reports, take a consistent snapshot at the START of the report run
// Store the snapshot in a separate reporting table updated at report time
// Report reads only from snapshot, not live tables
 
@Scheduled(cron = "0 0 1 * * *")  // Daily at 1am
public void generateDailyReportSnapshot() {
    Instant snapshotTime = Instant.now();
 
    // Take consistent snapshot
    List<OrderSummary> completedOrders = orderRepository
        .findByStatusAndCompletedBefore(OrderStatus.CONFIRMED, snapshotTime);
 
    // Store in reporting table
    reportingRepository.saveSnapshot(new DailySnapshot(snapshotTime, completedOrders));
}

@Transactional
public void reserveInventory(String productId, int quantity) {
    // PESSIMISTIC WRITE LOCK prevents concurrent reads
    InventoryItem item = inventoryRepository
        .findByProductIdWithLock(productId)
        .orElseThrow();
 
    if (item.getAvailableQuantity() < quantity) {
        throw new InsufficientStockException(productId);
    }
 
    item.setAvailableQuantity(item.getAvailableQuantity() - quantity);
    inventoryRepository.save(item);
    // Lock released when transaction commits
}

// Instead of: available = available - quantity (order-dependent)
// Use: available = total - reserved (always computable from total and reserved)
 
UPDATE inventory
SET reserved_quantity = reserved_quantity + ?
WHERE product_id = ?
  AND (total_quantity - reserved_quantity) >= ?  -- Check condition in UPDATE

// Order cannot be re-created while saga_in_progress = true
// Only set saga_in_progress = false after C1 commits
 
public void createOrder(PlaceOrderRequest request) {
    // Prevent re-order while saga is active
    Optional<Order> existingOrder = orderRepository
        .findPendingSagaForCustomer(request.customerId());
 
    if (existingOrder.isPresent()) {
        throw new DuplicateOrderException(
            "Existing saga in progress for customer: " + request.customerId()
        );
    }
 
    // Create new order with saga_in_progress = true
}

// WRONG - No idempotency check
@KafkaListener(topics = "payment.events.payment-processed")
public void handlePaymentProcessed(PaymentProcessedEvent event, Acknowledgment ack) {
    inventoryService.reserveInventory(event.orderId(), event.items());
    ack.acknowledge();
    // If ACK fails, message is re-delivered and inventory reserved TWICE
}
 
// CORRECT - Database unique constraint as idempotency guard
@KafkaListener(topics = "payment.events.payment-processed")
@Transactional
public void handlePaymentProcessed(PaymentProcessedEvent event, Acknowledgment ack) {
    // Unique constraint on order_id in reservations table prevents duplicate
    try {
        inventoryService.reserveInventory(event.orderId(), event.items());
    } catch (DataIntegrityViolationException e) {
        // Already processed - safe to acknowledge and continue
        log.info("Duplicate event ignored (idempotent): orderId={}", event.orderId());
    }
    ack.acknowledge();
}

// SAGA DESIGN CHECKLIST
// For EVERY step in your saga, answer these questions BEFORE writing code:
//
// 1. What does this step do? (Forward action)
// 2. What should happen if we need to undo it? (Compensation)
// 3. Can this step be undone? (Some cannot - e.g., "sent an SMS")
// 4. If it cannot be undone, can it be placed AFTER the pivot transaction? (Retriable only)
// 5. Is the compensation idempotent?
// 6. Can the compensation fail? If so, what is the fallback?
 
enum OrderSagaStep {
    // Step                Compensation            Notes
    CREATE_ORDER,         // -> CANCEL_ORDER        Idempotent via order status check
    PROCESS_PAYMENT,      // -> REFUND_PAYMENT      Idempotent via payment status check
    RESERVE_INVENTORY,    // -> RELEASE_INVENTORY   Idempotent via reservation status check
    CREATE_SHIPMENT,      // -> CANCEL_SHIPMENT     After pivot - prefer retry over compensation
    SEND_CONFIRMATION,    // NO compensation!       Cannot unsend email - place LAST (fire-and-forget)
    UPDATE_ANALYTICS      // NO compensation!       Eventually consistent, self-healing
}

// BETTER: Coarser steps that batch related changes
// Step 1: Update ALL preferences in Service B in ONE local transaction
// Step 2: Sync to analytics (async, fire-and-forget)
 
// Only need a saga if preferences span DIFFERENT databases in DIFFERENT services
// If email and notification preferences are in the SAME database -> just use @Transactional
 
@Service
public class PreferencesService {
 
    @Transactional
    public void updateAllPreferences(String customerId, PreferencesRequest request) {
        // All in ONE local transaction - NO saga needed!
        preferenceRepository.updateEmailPreference(customerId, request.email());
        preferenceRepository.updateNotificationPreference(customerId, request.notification());
        preferenceRepository.updateUIPreference(customerId, request.ui());
 
        // Async fire-and-forget to analytics (no compensation needed)
        eventPublisher.publishAsync(new PreferencesUpdatedEvent(customerId));
    }
}

// BAD: Compensation blocks waiting for synchronous response
@Transactional
public void compensateOrder(String orderId) {
    // Block waiting for payment service to confirm refund
    PaymentRefundResponse response = paymentServiceClient.refund(orderId)
        .get(30, TimeUnit.SECONDS);  // BLOCKS the compensation chain!
 
    // If payment service is down, this blocks for 30 seconds
    // Then throws TimeoutException
    // Compensation fails, saga stuck in COMPENSATING state forever
 
    orderRepository.updateStatus(orderId, OrderStatus.CANCELLED);
}

// GOOD: Compensation is event-driven, never blocks
// Compensation chain is driven by events, not synchronous calls
 
// Order Service publishes CompensationRequestedEvent
@Transactional
public void initiateCompensation(String orderId, String reason) {
    // Update status to CANCELLING (semantic lock for compensation in progress)
    Order order = orderRepository.findById(orderId).orElseThrow();
    order.setStatus(OrderStatus.CANCELLING);
    order.setSagaInProgress(true);
    orderRepository.save(order);
 
    // Publish compensation event - Payment Service will react asynchronously
    outboxEventRepository.save(OutboxEvent.of(
        KafkaTopics.COMPENSATION_ORDER,
        orderId,
        new CompensationRequestedEvent(orderId, reason)
    ));
 
    // Return immediately - do NOT wait for compensation to complete
    // Compensation completion will arrive via PaymentRefundedEvent
}

// Mark the pivot transaction explicitly in your design
// Everything before pivot: compensable
// Everything after pivot: either retriable or fire-and-forget
 
enum OrderSagaStep {
    CREATE_ORDER,        // Compensable: cancel order
    PROCESS_PAYMENT,     // Compensable: refund payment
    RESERVE_INVENTORY,   // PIVOT: No compensation after this succeeds
    CREATE_SHIPMENT,     // Post-pivot: Retry up to 5 times before manual intervention
    SEND_NOTIFICATION,   // Post-pivot: Fire-and-forget (after shipment confirmed)
    CONFIRM_ORDER        // Post-pivot: Retry
}
 
// In orchestrator: once pivot commits, no more compensation
private SagaStep getCompensationStart(SagaStep failedAt) {
    return switch (failedAt) {
        case CREATE_ORDER      -> null;                         // Nothing to compensate
        case PROCESS_PAYMENT   -> SagaStep.COMPENSATE_ORDER;   // Cancel order
        case RESERVE_INVENTORY -> SagaStep.COMPENSATE_PAYMENT; // Refund, then cancel
        case CREATE_SHIPMENT,
             SEND_NOTIFICATION,
             CONFIRM_ORDER     -> null;  // POST-PIVOT: No compensation, retry only
    };
}

# application.yml
spring:
  kafka:
    consumer:
      # After max retries, message goes to DLT automatically
      # Never configure infinite retries

// ALWAYS configure DLT via Spring Kafka error handler
@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> kafkaTemplate) {
    DeadLetterPublishingRecoverer dltRecoverer =
        new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, exception) -> {
                // Custom routing: send to named DLT topic
                return new TopicPartition(record.topic() + ".DLT", -1);
            }
        );
 
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
        dltRecoverer,
        new FixedBackOff(2000L, 3L)  // 3 retries, 2 second delay, THEN DLT
    );
 
    // Add monitoring when messages go to DLT
    errorHandler.setRetryListeners((record, ex, deliveryAttempt) ->
        log.warn("Retry attempt {} for topic={}, key={}, error={}",
            deliveryAttempt, record.topic(), record.key(), ex.getMessage())
    );
 
    return errorHandler;
}

// BAD: Compensation waits synchronously for each step
@Transactional
public void compensate(SagaInstance saga) {
    // Step 1: Call Payment Service synchronously
    paymentServiceClient.refund(saga.getOrderId());  // 500ms
 
    // Step 2: Call Inventory Service synchronously
    inventoryServiceClient.releaseReservation(saga.getOrderId());  // 300ms
 
    // Step 3: Update order status
    orderService.cancelOrder(saga.getOrderId());  // 100ms
 
    // Total: 900ms of blocking
    // If ANY service is slow, EVERYTHING blocks
    // If ANY service is down, ENTIRE compensation fails
}

// GOOD: Each compensation step publishes an event and returns immediately
// Next compensation step starts when event is consumed
 
// Payment Service: receives InventoryReservationFailedEvent
@KafkaListener(topics = KafkaTopics.INVENTORY_RESERVATION_FAILED)
@Transactional
public void handleInventoryFailed(InventoryReservationFailedEvent event, Acknowledgment ack) {
    // Do the refund (fast local operation + outbox write)
    paymentService.refund(event.orderId(), event.failureReason());
 
    // Refund triggers PaymentRefundedEvent -> Order Service cancels order
    // This is ASYNC - returns immediately
    ack.acknowledge();
}

// Add duplicate detection metrics
@KafkaListener(topics = KafkaTopics.PAYMENT_PROCESSED)
@Transactional
public void handlePaymentProcessed(PaymentProcessedEvent event, Acknowledgment ack) {
 
    // Check for duplicate
    if (processedEventRepository.existsByEventIdAndServiceId(
            event.eventId(), "inventory-service")) {
        log.warn("DUPLICATE_EVENT detected: eventId={}, topic=PAYMENT_PROCESSED",
            event.eventId());
        meterRegistry.counter("events.duplicates",
            "topic", "payment_processed").increment();
        ack.acknowledge();  // Acknowledge to prevent re-delivery
        return;
    }
 
    // Process event
    inventoryService.reserveInventory(event);
 
    // Mark as processed
    processedEventRepository.save(
        new ProcessedEvent(event.eventId(), "inventory-service", Instant.now())
    );
 
    ack.acknowledge();
}

// Use orderId as Kafka KEY to ensure all events for the same order
// go to the SAME partition (Kafka routes by key hash to partition)
 
kafkaTemplate.send(
    new ProducerRecord<>(
        topic,
        orderId,    // <-- KEY: ensures same partition for same order
        eventPayload
    )
);

// Check current saga state before processing
@KafkaListener(topics = KafkaTopics.SHIPMENT_CREATED)
@Transactional
public void handleShipmentCreated(ShipmentCreatedEvent event, Acknowledgment ack) {
    Order order = orderRepository.findById(event.orderId()).orElse(null);
 
    if (order == null || order.getStatus() != OrderStatus.INVENTORY_RESERVED) {
        // Event arrived too early - requeue with delay
        log.warn("Out-of-order event: ShipmentCreated but order status is {}",
            order == null ? "NOT_FOUND" : order.getStatus());
 
        // Option 1: Delay and retry (using retry topic)
        throw new OutOfOrderEventException("Order not ready for shipment confirmation");
 
        // Option 2: Store in a pending event table and retry when ready
    }
 
    // Process normally
    orderService.confirmOrder(event.orderId(), event.shipmentId());
    ack.acknowledge();
}

// PendingEventStore.java
@Service
@RequiredArgsConstructor
public class PendingEventStore {
 
    private final PendingEventRepository pendingEventRepository;
 
    /**
     * Store an event that arrived before its prerequisites were met.
     * A scheduler will retry these events periodically.
     */
    @Transactional
    public void storeForLater(String eventType, String orderId, String payload) {
        pendingEventRepository.save(PendingEvent.builder()
            .eventId(UUID.randomUUID().toString())
            .eventType(eventType)
            .orderId(orderId)
            .payload(payload)
            .retryAfter(Instant.now().plusSeconds(5))  // Retry after 5 seconds
            .maxRetries(10)
            .createdAt(Instant.now())
            .build());
    }
 
    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void retryPendingEvents() {
        List<PendingEvent> events = pendingEventRepository
            .findByRetryAfterBeforeOrderByCreatedAtAsc(Instant.now(),
                PageRequest.of(0, 50));
 
        for (PendingEvent event : events) {
            // Re-publish to original topic
            // ... re-publishing logic
 
            if (event.getRetryCount() >= event.getMaxRetries()) {
                event.setStatus("FAILED");
            } else {
                event.setRetryCount(event.getRetryCount() + 1);
                event.setRetryAfter(Instant.now().plusSeconds(
                    (long) Math.pow(2, event.getRetryCount()) * 5  // Exponential backoff
                ));
            }
            pendingEventRepository.save(event);
        }
    }
}

// CompensationRetryService.java
@Service
@RequiredArgsConstructor
@Slf4j
public class CompensationRetryService {
 
    private final SagaInstanceRepository sagaRepository;
    private final OrderSagaOrchestrator orchestrator;
    private final AlertService alertService;
 
    /**
     * Find sagas stuck in compensation and retry.
     */
    @Scheduled(fixedDelay = 30000)  // Every 30 seconds
    public void retryStuckCompensations() {
        List<SagaInstance> compensatingSagas = sagaRepository.findByStatus(
            SagaStatus.COMPENSATING
        );
 
        Instant stuckThreshold = Instant.now().minus(5, ChronoUnit.MINUTES);
 
        for (SagaInstance saga : compensatingSagas) {
            if (saga.getUpdatedAt().isBefore(stuckThreshold)) {
                if (saga.getRetryCount() < 20) {
                    log.warn("Retrying stuck compensation: sagaId={}, step={}, attempt={}",
                        saga.getSagaId(), saga.getCurrentStep(), saga.getRetryCount());
 
                    saga.setRetryCount(saga.getRetryCount() + 1);
                    sagaRepository.save(saga);
 
                    orchestrator.replayCurrentCompensationStep(saga);
                } else {
                    // Escalate to human
                    log.error("COMPENSATION PERMANENTLY STUCK: sagaId={}", saga.getSagaId());
                    alertService.sendPagerDutyAlert(
                        "CRITICAL: Manual intervention needed for saga compensation",
                        saga.getSagaId()
                    );
                    saga.setStatus(SagaStatus.FAILED);
                    sagaRepository.save(saga);
                }
            }
        }
    }
}

// SagaManualInterventionController.java
@RestController
@RequestMapping("/admin/sagas")
@RequiredArgsConstructor
public class SagaManualInterventionController {
 
    private final SagaManualInterventionService interventionService;
 
    // Force-complete a compensation manually (after verifying manually)
    @PostMapping("/{sagaId}/compensations/{step}/force-complete")
    @PreAuthorize("hasRole('SAGA_ADMIN')")
    public ResponseEntity<Void> forceCompleteCompensation(
            @PathVariable String sagaId,
            @PathVariable String step,
            @RequestBody ForceCompleteRequest request) {
        interventionService.forceCompleteCompensation(sagaId, step, request.reason());
        return ResponseEntity.ok().build();
    }
 
    // Force-fail a saga (mark as FAILED, stop all processing)
    @PostMapping("/{sagaId}/force-fail")
    @PreAuthorize("hasRole('SAGA_ADMIN')")
    public ResponseEntity<Void> forceFail(
            @PathVariable String sagaId,
            @RequestBody ForceFailRequest request) {
        interventionService.forceFail(sagaId, request.reason(), request.operatorId());
        return ResponseEntity.ok().build();
    }
}

// ALWAYS add fields, never remove or rename
// Use @JsonIgnoreProperties(ignoreUnknown = true) on all event classes
 
@JsonIgnoreProperties(ignoreUnknown = true)  // CRITICAL: ignore unknown fields
public record OrderCreatedEvent(
    String eventId,
    String sagaId,
    String orderId,
    String customerId,
    @JsonProperty("userId") String userId,  // Added in v2 (customerId still works)
    BigDecimal totalAmount,
    @JsonProperty("amount") BigDecimal amount,  // Added in v2 alias
    List<OrderItemDto> items,
    Instant occurredAt
) {
    // Getter that returns either customerId or userId (backward compatible)
    @JsonIgnore
    public String getEffectiveCustomerId() {
        return customerId != null ? customerId : userId;
    }
}

// Event Router: handles multiple versions of the same event type
@Component
@RequiredArgsConstructor
public class VersionedEventRouter {
 
    @KafkaListener(topics = KafkaTopics.ORDER_CREATED)
    public void handleOrderCreated(
            @Payload String rawPayload,
            @Header("eventVersion") String version,
            Acknowledgment ack) {
 
        switch (version) {
            case "1" -> handleV1OrderCreated(rawPayload);
            case "2" -> handleV2OrderCreated(rawPayload);
            default  -> {
                log.warn("Unknown event version {} - trying latest handler", version);
                handleV2OrderCreated(rawPayload);
            }
        }
        ack.acknowledge();
    }
}

// NEVER use hard-coded service addresses in saga compensations
// ALWAYS use event-driven communication (Kafka/SQS topics)
// Topics are stable; service addresses change
 
// BAD:
restTemplate.post("http://inventory-service-v1.internal/api/compensation", request);
 
// GOOD:
kafkaTemplate.send(CommandTopics.INVENTORY_SERVICE, orderId,
    new ReleaseInventoryCommand(sagaId, orderId));
// Inventory Service listens to this topic regardless of its URL

// Set SLA alerts for each step
// If Payment step takes > 10 seconds, alert immediately
 
@Component
public class SagaStepSlaMonitor {
 
    private static final Map<SagaStep, Duration> STEP_SLA_LIMITS = Map.of(
        SagaStep.CREATE_ORDER, Duration.ofSeconds(2),
        SagaStep.PROCESS_PAYMENT, Duration.ofSeconds(10),
        SagaStep.RESERVE_INVENTORY, Duration.ofSeconds(5),
        SagaStep.CREATE_SHIPMENT, Duration.ofSeconds(30)
    );
 
    public void checkStepSla(SagaStep step, Duration actualDuration) {
        Duration slaLimit = STEP_SLA_LIMITS.getOrDefault(step, Duration.ofSeconds(30));
 
        if (actualDuration.compareTo(slaLimit) > 0) {
            log.warn("SLA BREACH: step={}, actual={}ms, limit={}ms",
                step, actualDuration.toMillis(), slaLimit.toMillis());
            meterRegistry.counter("saga.sla.breach", "step", step.name()).increment();
        }
    }
}

// SagaFailureScenarioTests.java
// Test one failure scenario per test method
@SpringBootTest
@Testcontainers
class SagaFailureScenarioTests {
 
    @Test void testPaymentDeclined_ordersGetCancelled() {}
    @Test void testInventoryOutOfStock_paymentRefunded_orderCancelled() {}
    @Test void testShippingUnavailable_sagaRetries() {}
    @Test void testDuplicateOrderCreatedEvent_idempotent() {}
    @Test void testOutOfOrderEvents_handled() {}
    @Test void testSagaTimeout_compensated() {}
    @Test void testCompensationFails_alertSent_manualIntervention() {}
    @Test void testNetworkPartitionDuringPayment_sagaRecovers() {}
}

// Every log line, every metric, every trace should include sagaId
// This enables end-to-end saga tracing across all services
 
// In event handlers, extract sagaId and set in MDC
@KafkaListener(topics = "payment.events.payment-processed")
public void handle(PaymentProcessedEvent event, Acknowledgment ack) {
    MDC.put("sagaId", event.sagaId());
    MDC.put("orderId", event.orderId());
 
    try {
        // All log statements within will include sagaId and orderId
        inventoryService.reserveInventory(event);
        ack.acknowledge();
    } finally {
        MDC.remove("sagaId");
        MDC.remove("orderId");
    }
}

// BAD: Mixed business and compensation logic
@Service
public class PaymentService {
    public void processPayment(OrderCreatedEvent event) { /* ... */ }
 
    public void refundPayment(String orderId) {
        // Compensation mixed in the same service = tight coupling
    }
 
    public void partialRefund(String orderId, BigDecimal amount) { /* ... */ }
    public void reverseChargeback(String orderId) { /* ... */ }
}
 
// GOOD: Separate compensation handler
@Service
public class PaymentService {
    public PaymentResult processPayment(OrderCreatedEvent event) { /* ... */ }
}
 
@Component
public class PaymentCompensationHandler {
    private final PaymentService paymentService;
 
    public void handleCompensation(RefundPaymentCommand command) {
        paymentService.refund(command.orderId(), command.reason());
    }
}

// OrderSagaOrchestratorUnitTest.java
@ExtendWith(MockitoExtension.class)
class OrderSagaOrchestratorUnitTest {
 
    @Mock private SagaInstanceRepository sagaRepository;
    @Mock private KafkaTemplate<String, Object> kafkaTemplate;
 
    @InjectMocks private OrderSagaOrchestrator orchestrator;
 
    @Test
    @DisplayName("Payment success should transition to RESERVE_INVENTORY step")
    void paymentSuccess_shouldTransitionToInventoryStep() {
        SagaInstance saga = createSagaInStep(SagaStep.PROCESS_PAYMENT);
        SagaReplyEnvelope reply = SagaReplyEnvelope.success(saga.getSagaId(),
            "PROCESS_PAYMENT", "{\"paymentId\":\"PAY-001\"}");
 
        orchestrator.handleSagaReply(reply, mock(Acknowledgment.class));
 
        ArgumentCaptor<SagaInstance> sagaCaptor = ArgumentCaptor.forClass(SagaInstance.class);
        verify(sagaRepository).save(sagaCaptor.capture());
        assertThat(sagaCaptor.getValue().getCurrentStep())
            .isEqualTo(SagaStep.RESERVE_INVENTORY);
    }
 
    @Test
    @DisplayName("Inventory failure should trigger payment compensation")
    void inventoryFailure_shouldTriggerPaymentCompensation() {
        SagaInstance saga = createSagaInStep(SagaStep.RESERVE_INVENTORY);
        SagaReplyEnvelope reply = SagaReplyEnvelope.failure(
            saga.getSagaId(), "RESERVE_INVENTORY", "Out of stock", "INSUFFICIENT_STOCK"
        );
 
        orchestrator.handleSagaReply(reply, mock(Acknowledgment.class));
 
        // Verify: compensation command sent to payment service
        verify(kafkaTemplate).send(
            eq(CommandTopics.PAYMENT_SERVICE),
            eq(saga.getOrderId()),
            any(RefundPaymentCommand.class)
        );
    }
}