google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
Reliable Workflow Compensation in Java: Implementing Saga Pattern and Compensation Logic
awiskar acharya
November 17, 2025
10:33 am

Introduction

Workflow compensation logic is essential for maintaining data consistency in distributed systems where long-running business processes may fail partially. Compensation refers to the mechanism of undoing or compensating for completed steps when a subsequent step fails. This comprehensive guide explores compensation patterns, Saga implementations, and reliable rollback strategies in Java.

Core Compensation Concepts

1. Compensation Patterns

  • Saga Pattern: Distributed transaction pattern with compensating transactions
  • Compensating Transaction: Action that semantically reverses a previous action
  • Orchestration vs Choreography: Centralized vs decentralized coordination
  • Event Sourcing: Reconstructing state through event history

2. Compensation Strategies

  • Forward Recovery: Retry and continue
  • Backward Recovery: Compensate completed steps
  • Mixed Recovery: Combination of forward and backward recovery

Project Setup and Dependencies

1. 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> <!-- Database --> <dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <scope>runtime</scope> </dependency> <!-- JSON Processing --> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency> <!-- Resilience --> <dependency> <groupId>org.springframework.retry</groupId> <artifactId>spring-retry</artifactId> </dependency> <dependency> <groupId>io.github.resilience4j</groupId> <artifactId>resilience4j-spring-boot2</artifactId> </dependency> <!-- Testing --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.testcontainers</groupId> <artifactId>postgresql</artifactId> <scope>test</scope> </dependency> </dependencies>

2. Application Configuration

# application.yml spring: datasource: url: jdbc:postgresql://localhost:5432/workflow-compensation username: postgres password: password jpa: hibernate: ddl-auto: update properties: hibernate: dialect: org.hibernate.dialect.PostgreSQLDialect format_sql: true workflow: compensation: max-retry-attempts: 3 retry-delay-ms: 1000 saga-timeout-minutes: 30 logging: level: com.example.compensation: DEBUG org.springframework.retry: INFO

Core Domain Models

1. Workflow and Compensation Models

// Core workflow models @Entity @Table(name = "workflow_instances") @Data @NoArgsConstructor @AllArgsConstructor @Builder public class WorkflowInstance { @Id private String workflowId; @Enumerated(EnumType.STRING) private WorkflowStatus status; private String workflowType; private String businessKey; @ElementCollection @CollectionTable(name = "workflow_context", joinColumns = @JoinColumn(name = "workflow_id")) @MapKeyColumn(name = "context_key") @Column(name = "context_value") private Map<String, String> context; private LocalDateTime createdAt; private LocalDateTime updatedAt; private LocalDateTime completedAt; @Version private Long version; @PrePersist protected void onCreate() { this.createdAt = LocalDateTime.now(); this.updatedAt = LocalDateTime.now(); this.status = WorkflowStatus.PENDING; } @PreUpdate protected void onUpdate() { this.updatedAt = LocalDateTime.now(); if (this.status == WorkflowStatus.COMPLETED || this.status == WorkflowStatus.FAILED) { this.completedAt = LocalDateTime.now(); } } } @Entity @Table(name = "workflow_steps") @Data @NoArgsConstructor @AllArgsConstructor @Builder public class WorkflowStep { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; private String workflowId; private String stepId; private String stepType; @Enumerated(EnumType.STRING) private StepStatus status; @Column(columnDefinition = "TEXT") private String inputData; @Column(columnDefinition = "TEXT") private String outputData; @Column(columnDefinition = "TEXT") private String compensationData; private Integer attemptCount; private String errorMessage; private LocalDateTime startedAt; private LocalDateTime completedAt; private LocalDateTime compensatedAt; @Version private Long version; @PrePersist protected void onCreate() { this.startedAt = LocalDateTime.now(); this.status = StepStatus.PENDING; this.attemptCount = 0; } } @Entity @Table(name = "compensation_actions") @Data @NoArgsConstructor @AllArgsConstructor @Builder public class CompensationAction { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; private String workflowId; private String stepId; private String actionType; @Enumerated(EnumType.STRING) private CompensationStatus status; @Column(columnDefinition = "TEXT") private String actionData; private String errorMessage; private Integer attemptCount; private LocalDateTime createdAt; private LocalDateTime executedAt; @PrePersist protected void onCreate() { this.createdAt = LocalDateTime.now(); this.status = CompensationStatus.PENDING; this.attemptCount = 0; } } // Enums public enum WorkflowStatus { PENDING, RUNNING, COMPLETED, FAILED, COMPENSATING, COMPENSATED } public enum StepStatus { PENDING, RUNNING, COMPLETED, FAILED, COMPENSATED, SKIPPED } public enum CompensationStatus { PENDING, EXECUTED, FAILED, SKIPPED } // Compensation context @Data @NoArgsConstructor @AllArgsConstructor @Builder public class CompensationContext { private String workflowId; private String failedStepId; private String failureReason; private Map<String, Object> compensationData; private List<CompensatableStep> completedSteps; private LocalDateTime compensationStartedAt; public void addCompletedStep(CompensatableStep step) { if (completedSteps == null) { completedSteps = new ArrayList<>(); } completedSteps.add(step); } public List<CompensatableStep> getCompletedStepsInReverseOrder() { if (completedSteps == null) { return new ArrayList<>(); } List<CompensatableStep> reversed = new ArrayList<>(completedSteps); Collections.reverse(reversed); return reversed; } } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class CompensatableStep { private String stepId; private String stepType; private Object stepData; private Object compensationData; private LocalDateTime executedAt; }

2. Business Domain Models

// E-commerce domain models for demonstration @Data @NoArgsConstructor @AllArgsConstructor @Builder public class OrderRequest { private String orderId; private String customerId; private List<OrderItem> items; private PaymentInfo paymentInfo; private ShippingAddress shippingAddress; } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class OrderItem { private String productId; private String productName; private Integer quantity; private BigDecimal unitPrice; } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class PaymentInfo { private String paymentMethod; private String cardToken; private BigDecimal amount; private String currency; } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class ShippingAddress { private String addressLine1; private String addressLine2; private String city; private String state; private String zipCode; private String country; } // Step-specific data models @Data @NoArgsConstructor @AllArgsConstructor @Builder public class InventoryReservationData { private String reservationId; private List<InventoryItem> reservedItems; private LocalDateTime reservationExpiry; @Data @Builder public static class InventoryItem { private String productId; private Integer quantity; private String warehouseId; } } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class PaymentProcessingData { private String paymentId; private String transactionId; private BigDecimal amount; private String currency; private PaymentStatus status; public enum PaymentStatus { PENDING, COMPLETED, FAILED, REFUNDED } } @Data @NoArgsConstructor @AllArgsConstructor @Builder public class ShippingPreparationData { private String shipmentId; private String carrier; private String trackingNumber; private List<ShipmentItem> items; private LocalDateTime estimatedDelivery; }

Saga Orchestration Implementation

1. Saga Orchestrator

@Component @Slf4j public class SagaOrchestrator { private final WorkflowInstanceRepository workflowRepository; private final WorkflowStepRepository stepRepository; private final CompensationActionRepository compensationRepository; private final StepExecutor stepExecutor; private final CompensationExecutor compensationExecutor; private final ObjectMapper objectMapper; public SagaOrchestrator(WorkflowInstanceRepository workflowRepository, WorkflowStepRepository stepRepository, CompensationActionRepository compensationRepository, StepExecutor stepExecutor, CompensationExecutor compensationExecutor, ObjectMapper objectMapper) { this.workflowRepository = workflowRepository; this.stepRepository = stepRepository; this.compensationRepository = compensationRepository; this.stepExecutor = stepExecutor; this.compensationExecutor = compensationExecutor; this.objectMapper = objectMapper; } /** * Execute a saga workflow with compensation support */ @Transactional public WorkflowInstance executeSaga(String workflowType, String businessKey, List<WorkflowStepDefinition> steps) { String workflowId = generateWorkflowId(); WorkflowInstance workflow = WorkflowInstance.builder() .workflowId(workflowId) .workflowType(workflowType) .businessKey(businessKey) .status(WorkflowStatus.RUNNING) .context(new HashMap<>()) .build(); workflowRepository.save(workflow); log.info("Started saga workflow: {} for business key: {}", workflowId, businessKey); try { executeSteps(workflow, steps); workflow.setStatus(WorkflowStatus.COMPLETED); workflowRepository.save(workflow); log.info("Completed saga workflow: {}", workflowId); } catch (StepExecutionException e) { log.error("Saga workflow failed: {}", workflowId, e); compensateWorkflow(workflow, e.getFailedStepId(), e.getMessage()); } catch (Exception e) { log.error("Unexpected error in saga workflow: {}", workflowId, e); compensateWorkflow(workflow, null, "Unexpected error: " + e.getMessage()); } return workflow; } /** * Execute workflow steps sequentially */ private void executeSteps(WorkflowInstance workflow, List<WorkflowStepDefinition> steps) { List<CompensatableStep> completedSteps = new ArrayList<>(); for (WorkflowStepDefinition stepDef : steps) { WorkflowStep step = createWorkflowStep(workflow, stepDef); try { // Execute the step Object stepResult = stepExecutor.executeStep(stepDef.getStepType(), stepDef.getInputData()); // Record step completion step.setStatus(StepStatus.COMPLETED); step.setOutputData(objectMapper.writeValueAsString(stepResult)); step.setCompletedAt(LocalDateTime.now()); stepRepository.save(step); // Track for potential compensation CompensatableStep completedStep = CompensatableStep.builder() .stepId(stepDef.getStepId()) .stepType(stepDef.getStepType()) .stepData(stepDef.getInputData()) .compensationData(stepResult) // Store result for compensation .executedAt(LocalDateTime.now()) .build(); completedSteps.add(completedStep); log.debug("Completed step: {} in workflow: {}", stepDef.getStepId(), workflow.getWorkflowId()); } catch (StepExecutionException e) { // Step failed - mark it and throw to trigger compensation step.setStatus(StepStatus.FAILED); step.setErrorMessage(e.getMessage()); step.setCompletedAt(LocalDateTime.now()); stepRepository.save(step); throw new StepExecutionException(stepDef.getStepId(), e.getMessage(), e); } catch (Exception e) { // Unexpected error step.setStatus(StepStatus.FAILED); step.setErrorMessage("Unexpected error: " + e.getMessage()); step.setCompletedAt(LocalDateTime.now()); stepRepository.save(step); throw new StepExecutionException(stepDef.getStepId(), "Unexpected step execution error", e); } } // Store completed steps in workflow context for potential compensation if (!completedSteps.isEmpty()) { try { workflow.getContext().put("completedSteps", objectMapper.writeValueAsString(completedSteps)); workflowRepository.save(workflow); } catch (JsonProcessingException e) { log.warn("Failed to store completed steps in workflow context", e); } } } /** * Compensate completed steps when workflow fails */ @Transactional public void compensateWorkflow(WorkflowInstance workflow, String failedStepId, String failureReason) { log.info("Starting compensation for workflow: {}, failed at step: {}", workflow.getWorkflowId(), failedStepId); workflow.setStatus(WorkflowStatus.COMPENSATING); workflowRepository.save(workflow); CompensationContext context = CompensationContext.builder() .workflowId(workflow.getWorkflowId()) .failedStepId(failedStepId) .failureReason(failureReason) .compensationStartedAt(LocalDateTime.now()) .build(); try { // Retrieve completed steps from context List<CompensatableStep> completedSteps = extractCompletedSteps(workflow); context.setCompletedSteps(completedSteps); // Execute compensation in reverse order executeCompensation(context); workflow.setStatus(WorkflowStatus.COMPENSATED); workflowRepository.save(workflow); log.info("Successfully compensated workflow: {}", workflow.getWorkflowId()); } catch (CompensationException e) { log.error("Compensation failed for workflow: {}", workflow.getWorkflowId(), e); workflow.setStatus(WorkflowStatus.FAILED); workflowRepository.save(workflow); throw e; } catch (Exception e) { log.error("Unexpected error during compensation for workflow: {}", workflow.getWorkflowId(), e); workflow.setStatus(WorkflowStatus.FAILED); workflowRepository.save(workflow); throw new CompensationException("Unexpected compensation error", e); } } /** * Execute compensation actions in reverse order */ private void executeCompensation(CompensationContext context) { List<CompensatableStep> stepsToCompensate = context.getCompletedStepsInReverseOrder(); for (CompensatableStep step : stepsToCompensate) { CompensationAction action = createCompensationAction(context, step); try { compensationExecutor.executeCompensation(step.getStepType(), step.getCompensationData()); action.setStatus(CompensationStatus.EXECUTED); action.setExecutedAt(LocalDateTime.now()); compensationRepository.save(action); // Mark the step as compensated stepRepository.findByWorkflowIdAndStepId(context.getWorkflowId(), step.getStepId()) .ifPresent(workflowStep -> { workflowStep.setCompensatedAt(LocalDateTime.now()); workflowStep.setStatus(StepStatus.COMPENSATED); stepRepository.save(workflowStep); }); log.debug("Compensated step: {} in workflow: {}", step.getStepId(), context.getWorkflowId()); } catch (CompensationException e) { log.error("Compensation failed for step: {} in workflow: {}", step.getStepId(), context.getWorkflowId(), e); action.setStatus(CompensationStatus.FAILED); action.setErrorMessage(e.getMessage()); action.setAttemptCount(action.getAttemptCount() + 1); compensationRepository.save(action); // Continue with other compensations even if one fails // In some scenarios, you might want to stop or handle this differently } } } private WorkflowStep createWorkflowStep(WorkflowInstance workflow, WorkflowStepDefinition stepDef) { WorkflowStep step = WorkflowStep.builder() .workflowId(workflow.getWorkflowId()) .stepId(stepDef.getStepId()) .stepType(stepDef.getStepType()) .inputData(stepDef.getInputData()) .build(); return stepRepository.save(step); } private CompensationAction createCompensationAction(CompensationContext context, CompensatableStep step) { CompensationAction action = CompensationAction.builder() .workflowId(context.getWorkflowId()) .stepId(step.getStepId()) .actionType(step.getStepType() + "_COMPENSATION") .build(); try { action.setActionData(objectMapper.writeValueAsString( Map.of("stepData", step.getStepData(), "compensationData", step.getCompensationData()) )); } catch (JsonProcessingException e) { log.warn("Failed to serialize compensation action data", e); } return compensationRepository.save(action); } private List<CompensatableStep> extractCompletedSteps(WorkflowInstance workflow) { try { String completedStepsJson = workflow.getContext().get("completedSteps"); if (completedStepsJson != null) { return objectMapper.readValue(completedStepsJson, objectMapper.getTypeFactory().constructCollectionType(List.class, CompensatableStep.class)); } } catch (Exception e) { log.warn("Failed to extract completed steps from workflow context", e); } return new ArrayList<>(); } private String generateWorkflowId() { return "WF-" + UUID.randomUUID().toString(); } @Data @Builder public static class WorkflowStepDefinition { private String stepId; private String stepType; private String inputData; private String compensationHandler; } }

2. Step Execution Service

@Service @Slf4j public class StepExecutor { private final InventoryService inventoryService; private final PaymentService paymentService; private final ShippingService shippingService; private final NotificationService notificationService; private final ObjectMapper objectMapper; public StepExecutor(InventoryService inventoryService, PaymentService paymentService, ShippingService shippingService, NotificationService notificationService, ObjectMapper objectMapper) { this.inventoryService = inventoryService; this.paymentService = paymentService; this.shippingService = shippingService; this.notificationService = notificationService; this.objectMapper = objectMapper; } /** * Execute a workflow step based on step type */ public Object executeStep(String stepType, String inputData) { try { return switch (stepType) { case "VALIDATE_ORDER" -> validateOrder(inputData); case "RESERVE_INVENTORY" -> reserveInventory(inputData); case "PROCESS_PAYMENT" -> processPayment(inputData); case "PREPARE_SHIPMENT" -> prepareShipment(inputData); case "SEND_CONFIRMATION" -> sendConfirmation(inputData); default -> throw new StepExecutionException("Unknown step type: " + stepType); }; } catch (StepExecutionException e) { throw e; } catch (Exception e) { throw new StepExecutionException("Step execution failed: " + e.getMessage(), e); } } private Object validateOrder(String inputData) throws Exception { OrderRequest orderRequest = objectMapper.readValue(inputData, OrderRequest.class); // Validate order business rules if (orderRequest.getItems() == null || orderRequest.getItems().isEmpty()) { throw new StepExecutionException("Order must contain at least one item"); } if (orderRequest.getPaymentInfo() == null) { throw new StepExecutionException("Payment information is required"); } // Additional validation logic... log.info("Validated order: {}", orderRequest.getOrderId()); return Map.of("validated", true, "orderId", orderRequest.getOrderId()); } private Object reserveInventory(String inputData) throws Exception { OrderRequest orderRequest = objectMapper.readValue(inputData, OrderRequest.class); InventoryReservationData reservationData = inventoryService.reserveInventory( orderRequest.getOrderId(), orderRequest.getItems() ); log.info("Reserved inventory for order: {}, reservation: {}", orderRequest.getOrderId(), reservationData.getReservationId()); return reservationData; } private Object processPayment(String inputData) throws Exception { OrderRequest orderRequest = objectMapper.readValue(inputData, OrderRequest.class); PaymentProcessingData paymentData = paymentService.processPayment( orderRequest.getOrderId(), orderRequest.getCustomerId(), orderRequest.getPaymentInfo() ); if (paymentData.getStatus() != PaymentProcessingData.PaymentStatus.COMPLETED) { throw new StepExecutionException("Payment processing failed: " + paymentData.getStatus()); } log.info("Processed payment for order: {}, transaction: {}", orderRequest.getOrderId(), paymentData.getTransactionId()); return paymentData; } private Object prepareShipment(String inputData) throws Exception { OrderRequest orderRequest = objectMapper.readValue(inputData, OrderRequest.class); ShippingPreparationData shipmentData = shippingService.prepareShipment( orderRequest.getOrderId(), orderRequest.getItems(), orderRequest.getShippingAddress() ); log.info("Prepared shipment for order: {}, shipment: {}", orderRequest.getOrderId(), shipmentData.getShipmentId()); return shipmentData; } private Object sendConfirmation(String inputData) throws Exception { OrderRequest orderRequest = objectMapper.readValue(inputData, OrderRequest.class); notificationService.sendOrderConfirmation( orderRequest.getCustomerId(), orderRequest.getOrderId() ); log.info("Sent confirmation for order: {}", orderRequest.getOrderId()); return Map.of("confirmationSent", true); } }

3. Compensation Execution Service

@Service @Slf4j public class CompensationExecutor { private final InventoryService inventoryService; private final PaymentService paymentService; private final ShippingService shippingService; private final NotificationService notificationService; private final ObjectMapper objectMapper; public CompensationExecutor(InventoryService inventoryService, PaymentService paymentService, ShippingService shippingService, NotificationService notificationService, ObjectMapper objectMapper) { this.inventoryService = inventoryService; this.paymentService = paymentService; this.shippingService = shippingService; this.notificationService = notificationService; this.objectMapper = objectMapper; } /** * Execute compensation for a specific step type */ public void executeCompensation(String stepType, Object compensationData) { try { switch (stepType) { case "RESERVE_INVENTORY" -> compensateInventoryReservation(compensationData); case "PROCESS_PAYMENT" -> compensatePaymentProcessing(compensationData); case "PREPARE_SHIPMENT" -> compensateShipmentPreparation(compensationData); case "SEND_CONFIRMATION" -> compensateConfirmation(compensationData); default -> log.warn("No compensation defined for step type: {}", stepType); } } catch (CompensationException e) { throw e; } catch (Exception e) { throw new CompensationException("Compensation execution failed: " + e.getMessage(), e); } } private void compensateInventoryReservation(Object compensationData) throws Exception { InventoryReservationData reservationData = objectMapper.convertValue( compensationData, InventoryReservationData.class); inventoryService.releaseReservation(reservationData.getReservationId()); log.info("Compensated inventory reservation: {}", reservationData.getReservationId()); } private void compensatePaymentProcessing(Object compensationData) throws Exception { PaymentProcessingData paymentData = objectMapper.convertValue( compensationData, PaymentProcessingData.class); if (paymentData.getStatus() == PaymentProcessingData.PaymentStatus.COMPLETED) { paymentService.refundPayment(paymentData.getPaymentId()); log.info("Compensated payment: {}", paymentData.getPaymentId()); } else { log.info("Payment was not completed, no compensation needed: {}", paymentData.getPaymentId()); } } private void compensateShipmentPreparation(Object compensationData) throws Exception { ShippingPreparationData shipmentData = objectMapper.convertValue( compensationData, ShippingPreparationData.class); shippingService.cancelShipment(shipmentData.getShipmentId()); log.info("Compensated shipment: {}", shipmentData.getShipmentId()); } private void compensateConfirmation(Object compensationData) throws Exception { // For notification steps, compensation might involve sending a cancellation notification // or simply logging that the original notification should be ignored log.info("Compensated confirmation - sent cancellation notice"); // In practice, you might send a cancellation email // notificationService.sendOrderCancellation(...); } }

Business Service Implementations

1. Inventory Service with Compensation

@Service @Slf4j @Transactional public class InventoryService { private final InventoryRepository inventoryRepository; private final InventoryReservationRepository reservationRepository; public InventoryService(InventoryRepository inventoryRepository, InventoryReservationRepository reservationRepository) { this.inventoryRepository = inventoryRepository; this.reservationRepository = reservationRepository; } /** * Reserve inventory for an order */ public InventoryReservationData reserveInventory(String orderId, List<OrderItem> items) { String reservationId = "RES-" + UUID.randomUUID().toString(); List<InventoryReservationData.InventoryItem> reservedItems = new ArrayList<>(); for (OrderItem item : items) { InventoryItem inventory = inventoryRepository.findByProductId(item.getProductId()) .orElseThrow(() -> new InventoryException("Product not found: " + item.getProductId())); if (inventory.getAvailableQuantity() < item.getQuantity()) { throw new InventoryException("Insufficient inventory for product: " + item.getProductId()); } // Reserve the items inventory.setAvailableQuantity(inventory.getAvailableQuantity() - item.getQuantity()); inventory.setReservedQuantity(inventory.getReservedQuantity() + item.getQuantity()); inventoryRepository.save(inventory); // Record reservation InventoryReservationData.InventoryItem reservedItem = InventoryReservationData.InventoryItem.builder() .productId(item.getProductId()) .quantity(item.getQuantity()) .warehouseId(inventory.getWarehouseId()) .build(); reservedItems.add(reservedItem); log.debug("Reserved {} units of product {} for order {}", item.getQuantity(), item.getProductId(), orderId); } // Create reservation record InventoryReservation reservation = InventoryReservation.builder() .reservationId(reservationId) .orderId(orderId) .status(ReservationStatus.RESERVED) .expiresAt(LocalDateTime.now().plusHours(24)) // 24-hour reservation .createdAt(LocalDateTime.now()) .build(); reservationRepository.save(reservation); return InventoryReservationData.builder() .reservationId(reservationId) .reservedItems(reservedItems) .reservationExpiry(LocalDateTime.now().plusHours(24)) .build(); } /** * Release inventory reservation (compensation action) */ public void releaseReservation(String reservationId) { InventoryReservation reservation = reservationRepository.findById(reservationId) .orElseThrow(() -> new InventoryException("Reservation not found: " + reservationId)); if (reservation.getStatus() == ReservationStatus.RELEASED) { log.info("Reservation already released: {}", reservationId); return; } // Release reserved items back to available inventory // In a real implementation, you would track which items were reserved // For this example, we'll simulate the release reservation.setStatus(ReservationStatus.RELEASED); reservation.setReleasedAt(LocalDateTime.now()); reservationRepository.save(reservation); log.info("Released inventory reservation: {}", reservationId); } /** * Confirm reservation (convert reservation to permanent allocation) */ public void confirmReservation(String reservationId) { InventoryReservation reservation = reservationRepository.findById(reservationId) .orElseThrow(() -> new InventoryException("Reservation not found: " + reservationId)); reservation.setStatus(ReservationStatus.CONFIRMED); reservation.setConfirmedAt(LocalDateTime.now()); reservationRepository.save(reservation); log.info("Confirmed inventory reservation: {}", reservationId); } } @Entity @Table(name = "inventory_items") @Data @NoArgsConstructor @AllArgsConstructor @Builder public class InventoryItem { @Id private String productId; private String productName; private String warehouseId; private Integer availableQuantity; private Integer reservedQuantity; private Integer totalQuantity; @Version private Long version; } @Entity @Table(name = "inventory_reservations") @Data @NoArgsConstructor @AllArgsConstructor @Builder public class InventoryReservation { @Id private String reservationId; private String orderId; @Enumerated(EnumType.STRING) private ReservationStatus status; private LocalDateTime expiresAt; private LocalDateTime createdAt; private LocalDateTime releasedAt; private LocalDateTime confirmedAt; } public enum ReservationStatus { RESERVED, CONFIRMED, RELEASED, EXPIRED }

2. Payment Service with Compensation

@Service @Slf4j @Transactional public class PaymentService { private final PaymentRepository paymentRepository; private final PaymentGateway paymentGateway; public PaymentService(PaymentRepository paymentRepository, PaymentGateway paymentGateway) { this.paymentRepository = paymentRepository; this.paymentGateway = paymentGateway; } /** * Process payment for an order */ public PaymentProcessingData processPayment(String orderId, String customerId, PaymentInfo paymentInfo) { String paymentId = "PAY-" + UUID.randomUUID().toString(); try { // Process payment through payment gateway PaymentGatewayResponse response = paymentGateway.charge( paymentInfo.getAmount(), paymentInfo.getCurrency(), paymentInfo.getCardToken(), "Order: " + orderId ); // Record successful payment Payment payment = Payment.builder() .paymentId(paymentId) .orderId(orderId) .customerId(customerId) .amount(paymentInfo.getAmount()) .currency(paymentInfo.getCurrency()) .transactionId(response.getTransactionId()) .status(PaymentStatus.COMPLETED) .processedAt(LocalDateTime.now()) .build(); paymentRepository.save(payment); log.info("Processed payment for order: {}, transaction: {}", orderId, response.getTransactionId()); return PaymentProcessingData.builder() .paymentId(paymentId) .transactionId(response.getTransactionId()) .amount(paymentInfo.getAmount()) .currency(paymentInfo.getCurrency()) .status(PaymentProcessingData.PaymentStatus.COMPLETED) .build(); } catch (PaymentGatewayException e) { // Record failed payment attempt Payment payment = Payment.builder() .paymentId(paymentId) .orderId(orderId) .customerId(customerId) .amount(paymentInfo.getAmount()) .currency(paymentInfo.getCurrency()) .status(PaymentStatus.FAILED) .errorMessage(e.getMessage()) .processedAt(LocalDateTime.now()) .build(); paymentRepository.save(payment); throw new PaymentException("Payment processing failed: " + e.getMessage(), e); } } /** * Refund payment (compensation action) */ public void refundPayment(String paymentId) { Payment payment = paymentRepository.findById(paymentId) .orElseThrow(() -> new PaymentException("Payment not found: " + paymentId)); if (payment.getStatus() != PaymentStatus.COMPLETED) { throw new PaymentException("Cannot refund payment with status: " + payment.getStatus()); } try { // Process refund through payment gateway paymentGateway.refund(payment.getTransactionId(), payment.getAmount()); payment.setStatus(PaymentStatus.REFUNDED); payment.setRefundedAt(LocalDateTime.now()); paymentRepository.save(payment); log.info("Refunded payment: {}", paymentId); } catch (PaymentGatewayException e) { throw new PaymentException("Refund failed: " + e.getMessage(), e); } } }

Advanced Compensation Patterns

1. Retry and Circuit Breaker Patterns

@Service @Slf4j public class ResilientCompensationService { private final CompensationExecutor compensationExecutor; private final RetryTemplate retryTemplate; private final CircuitBreaker circuitBreaker; public ResilientCompensationService(CompensationExecutor compensationExecutor, RetryTemplate retryTemplate, CircuitBreakerRegistry circuitBreakerRegistry) { this.compensationExecutor = compensationExecutor; this.retryTemplate = retryTemplate; this.circuitBreaker = circuitBreakerRegistry.circuitBreaker("compensationService"); } /** * Execute compensation with retry and circuit breaker */ public void executeResilientCompensation(String stepType, Object compensationData) { try { circuitBreaker.executeRunnable(() -> retryTemplate.execute(context -> { log.info("Attempting compensation for step type: {}, attempt: {}", stepType, context.getRetryCount() + 1); compensationExecutor.executeCompensation(stepType, compensationData); return null; }) ); } catch (Exception e) { log.error("Compensation failed after retries for step type: {}", stepType, e); throw new CompensationException("Compensation failed after retries", e); } } } @Configuration @EnableRetry public class ResilienceConfig { @Bean public RetryTemplate retryTemplate() { return RetryTemplate.builder() .maxAttempts(3) .exponentialBackoff(1000, 2, 5000) .retryOn(RetryableCompensationException.class) .build(); } @Bean public Customizer<Resilience4JCircuitBreakerFactory> circuitBreakerFactoryCustomizer() { return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id) .circuitBreakerConfig(CircuitBreakerConfig.custom() .failureRateThreshold(50) .waitDurationInOpenState(Duration.ofSeconds(30)) .slidingWindowSize(10) .build()) .build()); } }

2. Compensation Monitoring and Metrics

@Component @Slf4j public class CompensationMetrics { private final MeterRegistry meterRegistry; private final Counter compensationAttemptCounter; private final Counter compensationSuccessCounter; private final Counter compensationFailureCounter; private final Timer compensationTimer; public CompensationMetrics(MeterRegistry meterRegistry) { this.meterRegistry = meterRegistry; this.compensationAttemptCounter = meterRegistry.counter("compensation.attempts"); this.compensationSuccessCounter = meterRegistry.counter("compensation.success"); this.compensationFailureCounter = meterRegistry.counter("compensation.failures"); this.compensationTimer = meterRegistry.timer("compensation.duration"); } public void recordCompensationAttempt(String stepType, String workflowId) { compensationAttemptCounter.increment(); Tags tags = Tags.of( Tag.of("step_type", stepType), Tag.of("workflow_id", workflowId) ); meterRegistry.counter("compensation.attempts.detailed", tags).increment(); log.debug("Compensation attempt - step: {}, workflow: {}", stepType, workflowId); } public void recordCompensationSuccess(String stepType, String workflowId, Duration duration) { compensationSuccessCounter.increment(); compensationTimer.record(duration); Tags tags = Tags.of( Tag.of("step_type", stepType), Tag.of("workflow_id", workflowId) ); meterRegistry.counter("compensation.success.detailed", tags).increment(); meterRegistry.timer("compensation.duration.detailed", tags).record(duration); log.info("Compensation successful - step: {}, workflow: {}, duration: {}ms", stepType, workflowId, duration.toMillis()); } public void recordCompensationFailure(String stepType, String workflowId, String errorType) { compensationFailureCounter.increment(); Tags tags = Tags.of( Tag.of("step_type", stepType), Tag.of("workflow_id", workflowId), Tag.of("error_type", errorType) ); meterRegistry.counter("compensation.failures.detailed", tags).increment(); log.warn("Compensation failed - step: {}, workflow: {}, error: {}", stepType, workflowId, errorType); } }

3. Compensation Recovery Service

@Service @Slf4j public class CompensationRecoveryService { private final CompensationActionRepository compensationRepository; private final WorkflowInstanceRepository workflowRepository; private final CompensationExecutor compensationExecutor; private final CompensationMetrics metrics; public CompensationRecoveryService(CompensationActionRepository compensationRepository, WorkflowInstanceRepository workflowRepository, CompensationExecutor compensationExecutor, CompensationMetrics metrics) { this.compensationRepository = compensationRepository; this.workflowRepository = workflowRepository; this.compensationExecutor = compensationExecutor; this.metrics = metrics; } /** * Recover failed compensations (e.g., after system restart) */ @Scheduled(fixedDelay = 300000) // Run every 5 minutes public void recoverFailedCompensations() { log.info("Starting compensation recovery process"); List<CompensationAction> failedActions = compensationRepository .findByStatusAndAttemptCountLessThan(CompensationStatus.FAILED, 3); for (CompensationAction action : failedActions) { try { recoverCompensationAction(action); } catch (Exception e) { log.error("Failed to recover compensation action: {}", action.getId(), e); } } log.info("Completed compensation recovery process, processed: {} actions", failedActions.size()); } /** * Recover stuck workflows in compensating state */ @Scheduled(fixedDelay = 600000) // Run every 10 minutes public void recoverStuckWorkflows() { log.info("Checking for stuck workflows"); LocalDateTime threshold = LocalDateTime.now().minusHours(1); List<WorkflowInstance> stuckWorkflows = workflowRepository .findByStatusAndUpdatedAtBefore(WorkflowStatus.COMPENSATING, threshold); for (WorkflowInstance workflow : stuckWorkflows) { log.warn("Found stuck workflow: {}, last updated: {}", workflow.getWorkflowId(), workflow.getUpdatedAt()); // Implement recovery logic based on business requirements // This might involve manual intervention, forced completion, or retry } } private void recoverCompensationAction(CompensationAction action) { try { metrics.recordCompensationAttempt(action.getActionType(), action.getWorkflowId()); // Parse compensation data and retry // Implementation would depend on how compensation data is stored compensationExecutor.executeCompensation( extractStepTypeFromActionType(action.getActionType()), parseCompensationData(action.getActionData()) ); action.setStatus(CompensationStatus.EXECUTED); action.setExecutedAt(LocalDateTime.now()); compensationRepository.save(action); metrics.recordCompensationSuccess(action.getActionType(), action.getWorkflowId(), Duration.between(action.getCreatedAt(), LocalDateTime.now())); } catch (Exception e) { log.error("Recovery failed for compensation action: {}", action.getId(), e); action.setAttemptCount(action.getAttemptCount() + 1); action.setErrorMessage("Recovery failed: " + e.getMessage()); compensationRepository.save(action); metrics.recordCompensationFailure(action.getActionType(), action.getWorkflowId(), e.getClass().getSimpleName()); } } private String extractStepTypeFromActionType(String actionType) { // Extract step type from action type (e.g., "RESERVE_INVENTORY_COMPENSATION" -> "RESERVE_INVENTORY") return actionType.replace("_COMPENSATION", ""); } private Object parseCompensationData(String actionData) { // Parse compensation data from JSON string // Implementation depends on data serialization format try { // This would use your preferred JSON library // return objectMapper.readValue(actionData, Map.class); return new Object(); // Placeholder } catch (Exception e) { throw new CompensationException("Failed to parse compensation data", e); } } }

REST API Controllers

1. Workflow Management API

@RestController @RequestMapping("/api/workflows") @Slf4j public class WorkflowController { private final SagaOrchestrator sagaOrchestrator; private final WorkflowInstanceRepository workflowRepository; private final WorkflowStepRepository stepRepository; private final ObjectMapper objectMapper; public WorkflowController(SagaOrchestrator sagaOrchestrator, WorkflowInstanceRepository workflowRepository, WorkflowStepRepository stepRepository, ObjectMapper objectMapper) { this.sagaOrchestrator = sagaOrchestrator; this.workflowRepository = workflowRepository; this.stepRepository = stepRepository; this.objectMapper = objectMapper; } @PostMapping("/orders") public ResponseEntity<WorkflowResponse> createOrder(@RequestBody OrderRequest orderRequest) { try { String workflowInput = objectMapper.writeValueAsString(orderRequest); List<SagaOrchestrator.WorkflowStepDefinition> steps = Arrays.asList( SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("validate-order") .stepType("VALIDATE_ORDER") .inputData(workflowInput) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("reserve-inventory") .stepType("RESERVE_INVENTORY") .inputData(workflowInput) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("process-payment") .stepType("PROCESS_PAYMENT") .inputData(workflowInput) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("prepare-shipment") .stepType("PREPARE_SHIPMENT") .inputData(workflowInput) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("send-confirmation") .stepType("SEND_CONFIRMATION") .inputData(workflowInput) .build() ); WorkflowInstance workflow = sagaOrchestrator.executeSaga( "ORDER_PROCESSING", orderRequest.getOrderId(), steps); return ResponseEntity.accepted() .body(WorkflowResponse.success(workflow)); } catch (Exception e) { log.error("Failed to create order workflow", e); return ResponseEntity.badRequest() .body(WorkflowResponse.error(e.getMessage())); } } @GetMapping("/{workflowId}") public ResponseEntity<WorkflowResponse> getWorkflow(@PathVariable String workflowId) { return workflowRepository.findById(workflowId) .map(workflow -> ResponseEntity.ok(WorkflowResponse.success(workflow))) .orElse(ResponseEntity.notFound().build()); } @GetMapping("/{workflowId}/steps") public ResponseEntity<List<WorkflowStep>> getWorkflowSteps(@PathVariable String workflowId) { List<WorkflowStep> steps = stepRepository.findByWorkflowId(workflowId); return ResponseEntity.ok(steps); } @PostMapping("/{workflowId}/compensate") public ResponseEntity<WorkflowResponse> manuallyCompensate(@PathVariable String workflowId) { try { WorkflowInstance workflow = workflowRepository.findById(workflowId) .orElseThrow(() -> new WorkflowNotFoundException("Workflow not found: " + workflowId)); sagaOrchestrator.compensateWorkflow(workflow, "MANUAL", "Manual compensation triggered"); return ResponseEntity.ok(WorkflowResponse.success(workflow)); } catch (Exception e) { log.error("Manual compensation failed for workflow: {}", workflowId, e); return ResponseEntity.badRequest() .body(WorkflowResponse.error("Compensation failed: " + e.getMessage())); } } @Data @Builder public static class WorkflowResponse { private boolean success; private String message; private WorkflowInstance workflow; private List<WorkflowStep> steps; public static WorkflowResponse success(WorkflowInstance workflow) { return WorkflowResponse.builder() .success(true) .message("Workflow processed successfully") .workflow(workflow) .build(); } public static WorkflowResponse error(String message) { return WorkflowResponse.builder() .success(false) .message(message) .build(); } } }

Testing Compensation Logic

1. Comprehensive Test Suite

@SpringBootTest @Testcontainers @ActiveProfiles("test") class WorkflowCompensationTest { @Container static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15"); @Autowired private SagaOrchestrator sagaOrchestrator; @Autowired private WorkflowInstanceRepository workflowRepository; @Autowired private WorkflowStepRepository stepRepository; @Autowired private CompensationActionRepository compensationRepository; @MockBean private PaymentService paymentService; @MockBean private InventoryService inventoryService; @DynamicPropertySource static void configureProperties(DynamicPropertyRegistry registry) { registry.add("spring.datasource.url", postgres::getJdbcUrl); registry.add("spring.datasource.username", postgres::getUsername); registry.add("spring.datasource.password", postgres::getPassword); } @Test void testSuccessfulWorkflowExecution() { // Given String orderId = "TEST-ORDER-123"; OrderRequest orderRequest = createTestOrderRequest(orderId); when(inventoryService.reserveInventory(anyString(), anyList())) .thenReturn(createInventoryReservation()); when(paymentService.processPayment(anyString(), anyString(), any())) .thenReturn(createPaymentProcessingData()); // When WorkflowInstance workflow = sagaOrchestrator.executeSaga( "ORDER_PROCESSING", orderId, createWorkflowSteps(orderRequest)); // Then assertNotNull(workflow); assertEquals(WorkflowStatus.COMPLETED, workflow.getStatus()); List<WorkflowStep> steps = stepRepository.findByWorkflowId(workflow.getWorkflowId()); assertEquals(5, steps.size()); assertTrue(steps.stream().allMatch(step -> step.getStatus() == StepStatus.COMPLETED)); } @Test void testWorkflowWithCompensation() { // Given String orderId = "TEST-ORDER-456"; OrderRequest orderRequest = createTestOrderRequest(orderId); when(inventoryService.reserveInventory(anyString(), anyList())) .thenReturn(createInventoryReservation()); when(paymentService.processPayment(anyString(), anyString(), any())) .thenThrow(new PaymentException("Payment gateway unavailable")); // When WorkflowInstance workflow = sagaOrchestrator.executeSaga( "ORDER_PROCESSING", orderId, createWorkflowSteps(orderRequest)); // Then assertNotNull(workflow); assertEquals(WorkflowStatus.COMPENSATED, workflow.getStatus()); List<WorkflowStep> steps = stepRepository.findByWorkflowId(workflow.getWorkflowId()); Optional<WorkflowStep> failedStep = steps.stream() .filter(step -> step.getStatus() == StepStatus.FAILED) .findFirst(); assertTrue(failedStep.isPresent()); assertEquals("process-payment", failedStep.get().getStepId()); List<CompensationAction> compensations = compensationRepository .findByWorkflowId(workflow.getWorkflowId()); assertFalse(compensations.isEmpty()); } @Test void testCompensationOrder() { // Given - Simulate a failed workflow with completed steps WorkflowInstance workflow = createFailedWorkflow(); List<WorkflowStep> completedSteps = createCompletedSteps(workflow.getWorkflowId()); // When sagaOrchestrator.compensateWorkflow(workflow, "process-payment", "Payment failed"); // Then WorkflowInstance updatedWorkflow = workflowRepository.findById(workflow.getWorkflowId()) .orElseThrow(); assertEquals(WorkflowStatus.COMPENSATED, updatedWorkflow.getStatus()); // Verify compensation actions were created in reverse order List<CompensationAction> compensations = compensationRepository .findByWorkflowId(workflow.getWorkflowId()); assertEquals(2, compensations.size()); // Two steps before the failure // Verify steps are marked as compensated List<WorkflowStep> steps = stepRepository.findByWorkflowId(workflow.getWorkflowId()); long compensatedSteps = steps.stream() .filter(step -> step.getStatus() == StepStatus.COMPENSATED) .count(); assertEquals(2, compensatedSteps); } private OrderRequest createTestOrderRequest(String orderId) { return OrderRequest.builder() .orderId(orderId) .customerId("test-customer") .items(List.of( OrderItem.builder() .productId("prod-1") .productName("Test Product") .quantity(2) .unitPrice(new BigDecimal("29.99")) .build() )) .paymentInfo(PaymentInfo.builder() .paymentMethod("CREDIT_CARD") .amount(new BigDecimal("59.98")) .currency("USD") .build()) .shippingAddress(ShippingAddress.builder() .city("Test City") .country("Test Country") .build()) .build(); } private List<SagaOrchestrator.WorkflowStepDefinition> createWorkflowSteps(OrderRequest orderRequest) { try { ObjectMapper objectMapper = new ObjectMapper(); String inputData = objectMapper.writeValueAsString(orderRequest); return Arrays.asList( SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("validate-order") .stepType("VALIDATE_ORDER") .inputData(inputData) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("reserve-inventory") .stepType("RESERVE_INVENTORY") .inputData(inputData) .build(), SagaOrchestrator.WorkflowStepDefinition.builder() .stepId("process-payment") .stepType("PROCESS_PAYMENT") .inputData(inputData) .build() ); } catch (Exception e) { throw new RuntimeException(e); } } private InventoryReservationData createInventoryReservation() { return InventoryReservationData.builder() .reservationId("TEST-RES-123") .reservedItems(new ArrayList<>()) .reservationExpiry(LocalDateTime.now().plusHours(24)) .build(); } private PaymentProcessingData createPaymentProcessingData() { return PaymentProcessingData.builder() .paymentId("TEST-PAY-123") .transactionId("TEST-TXN-123") .amount(new BigDecimal("59.98")) .currency("USD") .status(PaymentProcessingData.PaymentStatus.COMPLETED) .build(); } private WorkflowInstance createFailedWorkflow() { WorkflowInstance workflow = WorkflowInstance.builder() .workflowId("TEST-WF-123") .workflowType("ORDER_PROCESSING") .businessKey("TEST-ORDER") .status(WorkflowStatus.RUNNING) .context(new HashMap<>()) .build(); return workflowRepository.save(workflow); } private List<WorkflowStep> createCompletedSteps(String workflowId) { List<WorkflowStep> steps = Arrays.asList( WorkflowStep.builder() .workflowId(workflowId) .stepId("validate-order") .stepType("VALIDATE_ORDER") .status(StepStatus.COMPLETED) .completedAt(LocalDateTime.now()) .build(), WorkflowStep.builder() .workflowId(workflowId) .stepId("reserve-inventory") .stepType("RESERVE_INVENTORY") .status(StepStatus.COMPLETED) .completedAt(LocalDateTime.now()) .build() ); return stepRepository.saveAll(steps); } }

Conclusion

Workflow compensation logic is essential for building robust, reliable distributed systems. The Saga pattern with compensating transactions provides a powerful mechanism for maintaining data consistency across service boundaries.

Key Implementation Patterns:

  • Saga Orchestrator: Centralized coordination of workflow steps and compensation
  • Compensating Transactions: Semantic reversal of completed operations
  • Step Tracking: Comprehensive logging of workflow execution and compensation
  • Resilient Execution: Retry mechanisms and circuit breakers for reliability

Best Practices:

  • Idempotent Operations: Design steps and compensations to be safely retryable
  • Compensation Data: Store sufficient data to perform compensation actions
  • Monitoring: Comprehensive metrics and logging for observability
  • Recovery Mechanisms: Automated recovery of failed compensations
  • Testing: Thorough testing of both happy paths and failure scenarios

Use Cases:

  • E-commerce Order Processing: Inventory, payment, and shipping coordination
  • Financial Transactions: Multi-step money transfers and settlements
  • Travel Booking: Coordinating flights, hotels, and car rentals
  • Supply Chain Management: Inventory, manufacturing, and logistics coordination

By implementing proper workflow compensation logic, you can build systems that gracefully handle failures, maintain data consistency, and provide reliable business process execution even in the face of distributed system complexities.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper