MACRO NEPAL
Long-Running Workflows in Java: Building Resilient Business Processes
awiskar acharya
November 17, 2025
10:33 am

Long-running workflows are business processes that can span hours, days, or even months, requiring persistence, state management, and recovery mechanisms. They are essential for complex business operations like order processing, loan applications, and insurance claims.

Core Concepts

Characteristics of Long-Running Workflows

  • Durability: Survive process restarts and failures
  • State Management: Maintain process state across steps
  • Compensation: Handle failures with rollback logic
  • Resumability: Continue from last successful step
  • Monitoring: Track progress and performance

Implementation Approaches

1. State Machine Pattern with Persistence

Example 1: Order Processing Workflow

// Workflow State Enum
public enum WorkflowState {
CREATED,
VALIDATING,
PAYMENT_PROCESSING,
INVENTORY_RESERVATION,
SHIPPING,
COMPLETED,
CANCELLED,
FAILED
}
// Workflow Context
@Entity
@Table(name = "workflow_instances")
public class WorkflowInstance {
@Id
private String id;
@Enumerated(EnumType.STRING)
private WorkflowState currentState;
@Column(columnDefinition = "TEXT")
private String contextData; // JSON serialized workflow context
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
private LocalDateTime completedAt;
private int retryCount;
private String lastError;
@Version
private Long version;
// constructors, getters, setters
public WorkflowInstance() {
this.id = UUID.randomUUID().toString();
this.currentState = WorkflowState.CREATED;
this.createdAt = LocalDateTime.now();
this.updatedAt = LocalDateTime.now();
}
public WorkflowInstance(String id, WorkflowState state) {
this.id = id;
this.currentState = state;
this.createdAt = LocalDateTime.now();
this.updatedAt = LocalDateTime.now();
}
}
// Workflow Context Data
public class OrderWorkflowContext {
private String orderId;
private String customerId;
private BigDecimal amount;
private List<OrderItem> items;
private String paymentTransactionId;
private String shippingTrackingId;
private Map<String, Object> metadata = new HashMap<>();
// Compensation data
private boolean paymentCompensated = false;
private boolean inventoryCompensated = false;
// constructors, getters, setters
}
// Workflow Step Interface
public interface WorkflowStep {
String getName();
WorkflowState getTargetState();
WorkflowState getCompensationState();
WorkflowExecutionResult execute(WorkflowInstance instance, OrderWorkflowContext context);
WorkflowExecutionResult compensate(WorkflowInstance instance, OrderWorkflowContext context);
default boolean shouldExecute(WorkflowInstance instance, OrderWorkflowContext context) {
return true;
}
}
// Workflow Execution Result
public class WorkflowExecutionResult {
private final boolean success;
private final String message;
private final Map<String, Object> data;
private final Throwable error;
private WorkflowExecutionResult(boolean success, String message, 
Map<String, Object> data, Throwable error) {
this.success = success;
this.message = message;
this.data = data != null ? data : new HashMap<>();
this.error = error;
}
public static WorkflowExecutionResult success(String message) {
return new WorkflowExecutionResult(true, message, null, null);
}
public static WorkflowExecutionResult success(String message, Map<String, Object> data) {
return new WorkflowExecutionResult(true, message, data, null);
}
public static WorkflowExecutionResult failure(String message) {
return new WorkflowExecutionResult(false, message, null, null);
}
public static WorkflowExecutionResult failure(String message, Throwable error) {
return new WorkflowExecutionResult(false, message, null, error);
}
// getters
}
// Concrete Workflow Steps
@Component
@Slf4j
public class OrderValidationStep implements WorkflowStep {
private final OrderRepository orderRepository;
private final CustomerService customerService;
public OrderValidationStep(OrderRepository orderRepository, 
CustomerService customerService) {
this.orderRepository = orderRepository;
this.customerService = customerService;
}
@Override
public String getName() {
return "ORDER_VALIDATION";
}
@Override
public WorkflowState getTargetState() {
return WorkflowState.VALIDATING;
}
@Override
public WorkflowState getCompensationState() {
return WorkflowState.CREATED; // No compensation needed for validation
}
@Override
public WorkflowExecutionResult execute(WorkflowInstance instance, OrderWorkflowContext context) {
try {
log.info("Validating order: {}", context.getOrderId());
// 1. Validate order existence
Order order = orderRepository.findById(context.getOrderId())
.orElseThrow(() -> new WorkflowException("Order not found: " + context.getOrderId()));
// 2. Validate customer
Customer customer = customerService.getCustomer(context.getCustomerId());
if (customer == null) {
return WorkflowExecutionResult.failure("Customer not found: " + context.getCustomerId());
}
// 3. Validate order amount
if (order.getAmount().compareTo(BigDecimal.ZERO) <= 0) {
return WorkflowExecutionResult.failure("Invalid order amount: " + order.getAmount());
}
// 4. Check inventory availability (simplified)
if (!isInventoryAvailable(context.getItems())) {
return WorkflowExecutionResult.failure("Insufficient inventory");
}
log.info("Order validation successful: {}", context.getOrderId());
return WorkflowExecutionResult.success("Order validated successfully");
} catch (Exception e) {
log.error("Order validation failed for order: {}", context.getOrderId(), e);
return WorkflowExecutionResult.failure("Validation failed: " + e.getMessage(), e);
}
}
@Override
public WorkflowExecutionResult compensate(WorkflowInstance instance, OrderWorkflowContext context) {
// No compensation needed for validation
return WorkflowExecutionResult.success("No compensation needed for validation");
}
private boolean isInventoryAvailable(List<OrderItem> items) {
// Simplified inventory check
return items.stream().allMatch(item -> item.getQuantity() <= 100);
}
}
@Component
@Slf4j
public class PaymentProcessingStep implements WorkflowStep {
private final PaymentService paymentService;
public PaymentProcessingStep(PaymentService paymentService) {
this.paymentService = paymentService;
}
@Override
public String getName() {
return "PAYMENT_PROCESSING";
}
@Override
public WorkflowState getTargetState() {
return WorkflowState.PAYMENT_PROCESSING;
}
@Override
public WorkflowState getCompensationState() {
return WorkflowState.VALIDATING;
}
@Override
public WorkflowExecutionResult execute(WorkflowInstance instance, OrderWorkflowContext context) {
try {
log.info("Processing payment for order: {}", context.getOrderId());
PaymentRequest request = new PaymentRequest(
context.getOrderId(),
context.getCustomerId(),
context.getAmount(),
"USD"
);
PaymentResponse response = paymentService.processPayment(request);
if (response.isSuccess()) {
context.setPaymentTransactionId(response.getTransactionId());
log.info("Payment processed successfully: {}", response.getTransactionId());
Map<String, Object> resultData = new HashMap<>();
resultData.put("transactionId", response.getTransactionId());
resultData.put("paymentMethod", response.getPaymentMethod());
return WorkflowExecutionResult.success("Payment processed successfully", resultData);
} else {
return WorkflowExecutionResult.failure("Payment failed: " + response.getErrorMessage());
}
} catch (Exception e) {
log.error("Payment processing failed for order: {}", context.getOrderId(), e);
return WorkflowExecutionResult.failure("Payment processing failed: " + e.getMessage(), e);
}
}
@Override
public WorkflowExecutionResult compensate(WorkflowInstance instance, OrderWorkflowContext context) {
try {
if (context.getPaymentTransactionId() != null && !context.isPaymentCompensated()) {
log.info("Compensating payment for transaction: {}", context.getPaymentTransactionId());
boolean refunded = paymentService.refundPayment(context.getPaymentTransactionId());
if (refunded) {
context.setPaymentCompensated(true);
return WorkflowExecutionResult.success("Payment refunded successfully");
} else {
return WorkflowExecutionResult.failure("Payment refund failed");
}
}
return WorkflowExecutionResult.success("No payment to compensate");
} catch (Exception e) {
log.error("Payment compensation failed for transaction: {}", 
context.getPaymentTransactionId(), e);
return WorkflowExecutionResult.failure("Payment compensation failed: " + e.getMessage(), e);
}
}
}
@Component
@Slf4j
public class InventoryReservationStep implements WorkflowStep {
private final InventoryService inventoryService;
public InventoryReservationStep(InventoryService inventoryService) {
this.inventoryService = inventoryService;
}
@Override
public String getName() {
return "INVENTORY_RESERVATION";
}
@Override
public WorkflowState getTargetState() {
return WorkflowState.INVENTORY_RESERVATION;
}
@Override
public WorkflowState getCompensationState() {
return WorkflowState.PAYMENT_PROCESSING;
}
@Override
public WorkflowExecutionResult execute(WorkflowInstance instance, OrderWorkflowContext context) {
try {
log.info("Reserving inventory for order: {}", context.getOrderId());
List<InventoryReservation> reservations = context.getItems().stream()
.map(item -> new InventoryReservation(
item.getProductId(),
item.getQuantity(),
context.getOrderId()
))
.collect(Collectors.toList());
boolean reserved = inventoryService.reserveItems(reservations);
if (reserved) {
log.info("Inventory reserved successfully for order: {}", context.getOrderId());
return WorkflowExecutionResult.success("Inventory reserved successfully");
} else {
return WorkflowExecutionResult.failure("Inventory reservation failed");
}
} catch (Exception e) {
log.error("Inventory reservation failed for order: {}", context.getOrderId(), e);
return WorkflowExecutionResult.failure("Inventory reservation failed: " + e.getMessage(), e);
}
}
@Override
public WorkflowExecutionResult compensate(WorkflowInstance instance, OrderWorkflowContext context) {
try {
if (!context.isInventoryCompensated()) {
log.info("Compensating inventory for order: {}", context.getOrderId());
boolean released = inventoryService.releaseReservation(context.getOrderId());
if (released) {
context.setInventoryCompensated(true);
return WorkflowExecutionResult.success("Inventory reservation released");
} else {
return WorkflowExecutionResult.failure("Inventory release failed");
}
}
return WorkflowExecutionResult.success("No inventory to compensate");
} catch (Exception e) {
log.error("Inventory compensation failed for order: {}", context.getOrderId(), e);
return WorkflowExecutionResult.failure("Inventory compensation failed: " + e.getMessage(), e);
}
}
}
// Workflow Engine
@Component
@Slf4j
public class WorkflowEngine {
private final WorkflowInstanceRepository instanceRepository;
private final List<WorkflowStep> workflowSteps;
private final ObjectMapper objectMapper;
private final TaskScheduler taskScheduler;
public WorkflowEngine(WorkflowInstanceRepository instanceRepository,
List<WorkflowStep> workflowSteps,
ObjectMapper objectMapper,
TaskScheduler taskScheduler) {
this.instanceRepository = instanceRepository;
this.workflowSteps = workflowSteps;
this.objectMapper = objectMapper;
this.taskScheduler = taskScheduler;
}
@Transactional
public WorkflowInstance startWorkflow(String workflowType, Object context) {
try {
WorkflowInstance instance = new WorkflowInstance();
instance.setCurrentState(WorkflowState.CREATED);
String contextData = objectMapper.writeValueAsString(context);
instance.setContextData(contextData);
WorkflowInstance savedInstance = instanceRepository.save(instance);
log.info("Started workflow instance: {}", savedInstance.getId());
// Schedule execution
scheduleWorkflowExecution(savedInstance.getId());
return savedInstance;
} catch (Exception e) {
throw new WorkflowException("Failed to start workflow", e);
}
}
public void scheduleWorkflowExecution(String instanceId) {
taskScheduler.schedule(() -> {
try {
executeWorkflowStep(instanceId);
} catch (Exception e) {
log.error("Failed to execute workflow step for instance: {}", instanceId, e);
}
}, Instant.now().plusSeconds(1));
}
@Transactional
public void executeWorkflowStep(String instanceId) {
WorkflowInstance instance = instanceRepository.findById(instanceId)
.orElseThrow(() -> new WorkflowException("Workflow instance not found: " + instanceId));
try {
OrderWorkflowContext context = objectMapper.readValue(
instance.getContextData(), OrderWorkflowContext.class);
WorkflowStep nextStep = findNextStep(instance.getCurrentState());
if (nextStep == null) {
log.info("Workflow completed: {}", instanceId);
instance.setCurrentState(WorkflowState.COMPLETED);
instance.setCompletedAt(LocalDateTime.now());
instanceRepository.save(instance);
return;
}
if (nextStep.shouldExecute(instance, context)) {
log.info("Executing step: {} for workflow: {}", nextStep.getName(), instanceId);
instance.setCurrentState(nextStep.getTargetState());
instance.setUpdatedAt(LocalDateTime.now());
WorkflowExecutionResult result = nextStep.execute(instance, context);
if (result.isSuccess()) {
// Update context with result data
context.getMetadata().putAll(result.getData());
instance.setContextData(objectMapper.writeValueAsString(context));
instance.setRetryCount(0);
instance.setLastError(null);
instanceRepository.save(instance);
log.info("Step {} completed successfully for workflow: {}", 
nextStep.getName(), instanceId);
// Schedule next step
scheduleWorkflowExecution(instanceId);
} else {
handleStepFailure(instance, context, nextStep, result);
}
} else {
log.info("Skipping step: {} for workflow: {}", nextStep.getName(), instanceId);
scheduleWorkflowExecution(instanceId);
}
} catch (Exception e) {
log.error("Workflow execution failed for instance: {}", instanceId, e);
handleExecutionError(instance, e);
}
}
private WorkflowStep findNextStep(WorkflowState currentState) {
return workflowSteps.stream()
.filter(step -> step.getTargetState().ordinal() > currentState.ordinal())
.min(Comparator.comparing(step -> step.getTargetState().ordinal()))
.orElse(null);
}
private void handleStepFailure(WorkflowInstance instance, OrderWorkflowContext context,
WorkflowStep failedStep, WorkflowExecutionResult result) {
try {
instance.setRetryCount(instance.getRetryCount() + 1);
instance.setLastError(result.getMessage());
instance.setUpdatedAt(LocalDateTime.now());
if (instance.getRetryCount() >= 3) {
log.error("Step {} failed after {} retries for workflow: {}", 
failedStep.getName(), instance.getRetryCount(), instance.getId());
// Start compensation
startCompensation(instance, context, failedStep);
} else {
// Retry the same step
log.info("Scheduling retry {} for step {} in workflow: {}", 
instance.getRetryCount(), failedStep.getName(), instance.getId());
instanceRepository.save(instance);
// Schedule retry with exponential backoff
long delaySeconds = (long) Math.pow(2, instance.getRetryCount()) * 5;
taskScheduler.schedule(() -> executeWorkflowStep(instance.getId()),
Instant.now().plusSeconds(delaySeconds));
}
} catch (Exception e) {
log.error("Error handling step failure for workflow: {}", instance.getId(), e);
handleExecutionError(instance, e);
}
}
private void startCompensation(WorkflowInstance instance, OrderWorkflowContext context,
WorkflowStep failedStep) {
try {
log.info("Starting compensation for workflow: {}", instance.getId());
// Execute compensation for failed step and previous steps
List<WorkflowStep> stepsToCompensate = workflowSteps.stream()
.filter(step -> step.getTargetState().ordinal() >= failedStep.getCompensationState().ordinal()
&& step.getTargetState().ordinal() <= failedStep.getTargetState().ordinal())
.sorted((s1, s2) -> Integer.compare(s2.getTargetState().ordinal(), s1.getTargetState().ordinal()))
.collect(Collectors.toList());
for (WorkflowStep step : stepsToCompensate) {
log.info("Compensating step: {} for workflow: {}", step.getName(), instance.getId());
WorkflowExecutionResult compensationResult = step.compensate(instance, context);
if (!compensationResult.isSuccess()) {
log.error("Compensation failed for step: {} in workflow: {}", 
step.getName(), instance.getId());
// Continue with other compensations despite failures
}
}
instance.setCurrentState(WorkflowState.FAILED);
instance.setContextData(objectMapper.writeValueAsString(context));
instanceRepository.save(instance);
log.info("Compensation completed for workflow: {}", instance.getId());
} catch (Exception e) {
log.error("Compensation process failed for workflow: {}", instance.getId(), e);
instance.setCurrentState(WorkflowState.FAILED);
instanceRepository.save(instance);
}
}
private void handleExecutionError(WorkflowInstance instance, Exception error) {
instance.setCurrentState(WorkflowState.FAILED);
instance.setLastError(error.getMessage());
instance.setUpdatedAt(LocalDateTime.now());
instanceRepository.save(instance);
}
@Transactional
public void resumeWorkflow(String instanceId) {
WorkflowInstance instance = instanceRepository.findById(instanceId)
.orElseThrow(() -> new WorkflowException("Workflow instance not found: " + instanceId));
if (instance.getCurrentState() == WorkflowState.FAILED) {
log.info("Resuming failed workflow: {}", instanceId);
instance.setCurrentState(WorkflowState.CREATED);
instance.setRetryCount(0);
instance.setLastError(null);
instanceRepository.save(instance);
scheduleWorkflowExecution(instanceId);
}
}
public WorkflowInstance getWorkflowStatus(String instanceId) {
return instanceRepository.findById(instanceId)
.orElseThrow(() -> new WorkflowException("Workflow instance not found: " + instanceId));
}
}
// Custom Exceptions
class WorkflowException extends RuntimeException {
public WorkflowException(String message) {
super(message);
}
public WorkflowException(String message, Throwable cause) {
super(message, cause);
}
}

2. Saga Pattern for Distributed Transactions

Example 2: Saga Pattern Implementation

// Saga Definition
public class OrderSaga {
private String sagaId;
private String orderId;
private SagaState state;
private List<SagaStep> steps;
private int currentStep;
private Map<String, Object> context;
private LocalDateTime createdAt;
private LocalDateTime updatedAt;
// constructors, getters, setters
}
// Saga Step
public interface SagaStep {
String getName();
SagaExecutionResult execute(Saga saga);
SagaExecutionResult compensate(Saga saga);
}
// Saga Coordinator
@Component
@Slf4j
public class SagaCoordinator {
private final SagaRepository sagaRepository;
private final Map<String, SagaStep> sagaSteps;
private final ObjectMapper objectMapper;
public SagaCoordinator(SagaRepository sagaRepository,
List<SagaStep> steps,
ObjectMapper objectMapper) {
this.sagaRepository = sagaRepository;
this.objectMapper = objectMapper;
this.sagaSteps = steps.stream()
.collect(Collectors.toMap(SagaStep::getName, Function.identity()));
}
@Transactional
public String startSaga(String sagaType, Map<String, Object> initialContext) {
Saga saga = new Saga();
saga.setSagaId(UUID.randomUUID().toString());
saga.setState(SagaState.STARTED);
saga.setContext(initialContext);
saga.setCreatedAt(LocalDateTime.now());
saga.setUpdatedAt(LocalDateTime.now());
// Define saga steps based on type
saga.setSteps(defineSagaSteps(sagaType));
saga.setCurrentStep(0);
Saga savedSaga = sagaRepository.save(saga);
executeNextStep(savedSaga);
return savedSaga.getSagaId();
}
private List<SagaStep> defineSagaSteps(String sagaType) {
// Return steps based on saga type
return List.of(
sagaSteps.get("validateOrder"),
sagaSteps.get("processPayment"),
sagaSteps.get("reserveInventory"),
sagaSteps.get("shipOrder")
);
}
public void executeNextStep(Saga saga) {
if (saga.getCurrentStep() >= saga.getSteps().size()) {
saga.setState(SagaState.COMPLETED);
sagaRepository.save(saga);
return;
}
SagaStep currentStep = saga.getSteps().get(saga.getCurrentStep());
try {
SagaExecutionResult result = currentStep.execute(saga);
if (result.isSuccess()) {
saga.setCurrentStep(saga.getCurrentStep() + 1);
saga.setUpdatedAt(LocalDateTime.now());
sagaRepository.save(saga);
executeNextStep(saga);
} else {
startCompensation(saga);
}
} catch (Exception e) {
log.error("Saga step execution failed: {}", currentStep.getName(), e);
startCompensation(saga);
}
}
private void startCompensation(Saga saga) {
saga.setState(SagaState.COMPENSATING);
sagaRepository.save(saga);
// Execute compensation in reverse order
for (int i = saga.getCurrentStep() - 1; i >= 0; i--) {
SagaStep step = saga.getSteps().get(i);
try {
step.compensate(saga);
} catch (Exception e) {
log.error("Compensation failed for step: {}", step.getName(), e);
}
}
saga.setState(SagaState.FAILED);
sagaRepository.save(saga);
}
}

3. Temporal.io for Workflow Orchestration

Example 3: Temporal Workflow Implementation

// Temporal Workflow Interface
@WorkflowInterface
public interface OrderWorkflow {
@WorkflowMethod
void processOrder(Order order);
@SignalMethod
void updatePaymentStatus(PaymentStatus status);
@SignalMethod
void cancelOrder();
@QueryMethod
OrderStatus getOrderStatus();
}
// Temporal Workflow Implementation
public class OrderWorkflowImpl implements OrderWorkflow {
private final OrderActivities activities = Workflow.newActivityStub(OrderActivities.class);
private OrderStatus status = OrderStatus.CREATED;
private boolean cancelled = false;
@Override
public void processOrder(Order order) {
try {
// Step 1: Validate order
activities.validateOrder(order);
status = OrderStatus.VALIDATED;
// Wait for payment
Workflow.await(() -> status == OrderStatus.PAYMENT_RECEIVED || cancelled);
if (cancelled) {
activities.cancelOrder(order);
status = OrderStatus.CANCELLED;
return;
}
// Step 2: Reserve inventory
activities.reserveInventory(order);
status = OrderStatus.INVENTORY_RESERVED;
// Step 3: Ship order
activities.shipOrder(order);
status = OrderStatus.SHIPPED;
// Step 4: Send confirmation
activities.sendConfirmation(order);
status = OrderStatus.COMPLETED;
} catch (Exception e) {
activities.compensate(order);
status = OrderStatus.FAILED;
throw e;
}
}
@Override
public void updatePaymentStatus(PaymentStatus paymentStatus) {
if (paymentStatus == PaymentStatus.COMPLETED) {
this.status = OrderStatus.PAYMENT_RECEIVED;
}
}
@Override
public void cancelOrder() {
this.cancelled = true;
}
@Override
public OrderStatus getOrderStatus() {
return status;
}
}
// Temporal Activities
public interface OrderActivities {
@ActivityMethod
void validateOrder(Order order);
@ActivityMethod
void reserveInventory(Order order);
@ActivityMethod
void shipOrder(Order order);
@ActivityMethod
void sendConfirmation(Order order);
@ActivityMethod
void cancelOrder(Order order);
@ActivityMethod
void compensate(Order order);
}

4. Event-Driven Workflows with State Machines

Example 4: Event-Driven Workflow with Spring State Machine

// Configuration
@Configuration
@EnableStateMachine
public class StateMachineConfig extends StateMachineConfigurerAdapter<WorkflowState, WorkflowEvent> {
@Override
public void configure(StateMachineStateConfigurer<WorkflowState, WorkflowEvent> states)
throws Exception {
states
.withStates()
.initial(WorkflowState.CREATED)
.state(WorkflowState.VALIDATING)
.state(WorkflowState.PAYMENT_PROCESSING)
.state(WorkflowState.INVENTORY_RESERVATION)
.state(WorkflowState.SHIPPING)
.end(WorkflowState.COMPLETED)
.end(WorkflowState.CANCELLED)
.end(WorkflowState.FAILED);
}
@Override
public void configure(StateMachineTransitionConfigurer<WorkflowState, WorkflowEvent> transitions)
throws Exception {
transitions
.withExternal()
.source(WorkflowState.CREATED).target(WorkflowState.VALIDATING)
.event(WorkflowEvent.START_PROCESSING)
.action(validationAction())
.and()
.withExternal()
.source(WorkflowState.VALIDATING).target(WorkflowState.PAYMENT_PROCESSING)
.event(WorkflowEvent.VALIDATION_SUCCESS)
.action(paymentAction())
.and()
.withExternal()
.source(WorkflowState.VALIDATING).target(WorkflowState.FAILED)
.event(WorkflowEvent.VALIDATION_FAILED)
.and()
.withExternal()
.source(WorkflowState.PAYMENT_PROCESSING).target(WorkflowState.INVENTORY_RESERVATION)
.event(WorkflowEvent.PAYMENT_SUCCESS)
.action(inventoryAction())
.and()
.withExternal()
.source(WorkflowState.PAYMENT_PROCESSING).target(WorkflowState.FAILED)
.event(WorkflowEvent.PAYMENT_FAILED);
}
@Bean
public Action<WorkflowState, WorkflowEvent> validationAction() {
return context -> {
OrderWorkflowContext workflowContext = context.getExtendedState()
.get("context", OrderWorkflowContext.class);
// Execute validation logic
};
}
// Other action beans...
}
// Workflow Service
@Service
@Slf4j
public class StateMachineWorkflowService {
private final StateMachineFactory<WorkflowState, WorkflowEvent> stateMachineFactory;
private final StateMachinePersist<WorkflowState, WorkflowEvent, String> persister;
public StateMachineWorkflowService(StateMachineFactory<WorkflowState, WorkflowEvent> stateMachineFactory,
StateMachinePersist<WorkflowState, WorkflowEvent, String> persister) {
this.stateMachineFactory = stateMachineFactory;
this.persister = persister;
}
public String startWorkflow(OrderWorkflowContext context) {
String workflowId = UUID.randomUUID().toString();
StateMachine<WorkflowState, WorkflowEvent> stateMachine = stateMachineFactory.getStateMachine(workflowId);
stateMachine.getExtendedState().getVariables().put("context", context);
stateMachine.start();
stateMachine.sendEvent(WorkflowEvent.START_PROCESSING);
return workflowId;
}
public void handleEvent(String workflowId, WorkflowEvent event) {
StateMachine<WorkflowState, WorkflowEvent> stateMachine = stateMachineFactory.getStateMachine(workflowId);
stateMachine.sendEvent(event);
}
public WorkflowState getWorkflowState(String workflowId) {
StateMachine<WorkflowState, WorkflowEvent> stateMachine = stateMachineFactory.getStateMachine(workflowId);
return stateMachine.getState().getId();
}
}

Advanced Patterns

5. Workflow Recovery and Monitoring

Example 5: Workflow Recovery Service

@Service
@Slf4j
public class WorkflowRecoveryService {
private final WorkflowInstanceRepository instanceRepository;
private final WorkflowEngine workflowEngine;
private final MeterRegistry meterRegistry;
public WorkflowRecoveryService(WorkflowInstanceRepository instanceRepository,
WorkflowEngine workflowEngine,
MeterRegistry meterRegistry) {
this.instanceRepository = instanceRepository;
this.workflowEngine = workflowEngine;
this.meterRegistry = meterRegistry;
}
@Scheduled(fixedRate = 30000) // Run every 30 seconds
public void recoverStalledWorkflows() {
log.info("Starting workflow recovery process");
LocalDateTime cutoffTime = LocalDateTime.now().minusMinutes(5);
List<WorkflowInstance> stalledWorkflows = instanceRepository
.findByCurrentStateNotInAndUpdatedAtBefore(
List.of(WorkflowState.COMPLETED, WorkflowState.FAILED, WorkflowState.CANCELLED),
cutoffTime
);
log.info("Found {} stalled workflows", stalledWorkflows.size());
for (WorkflowInstance instance : stalledWorkflows) {
try {
log.info("Recovering stalled workflow: {}", instance.getId());
workflowEngine.scheduleWorkflowExecution(instance.getId());
meterRegistry.counter("workflow.recovery.attempted").increment();
} catch (Exception e) {
log.error("Failed to recover workflow: {}", instance.getId(), e);
meterRegistry.counter("workflow.recovery.failed").increment();
}
}
}
@Scheduled(cron = "0 0 2 * * ?") // Daily at 2 AM
public void cleanupOldWorkflows() {
log.info("Starting workflow cleanup");
LocalDateTime cutoffTime = LocalDateTime.now().minusDays(30);
int deletedCount = instanceRepository.deleteByCompletedAtBefore(cutoffTime);
log.info("Cleaned up {} old workflow instances", deletedCount);
meterRegistry.counter("workflow.cleanup.executed").increment();
}
}
// Monitoring Service
@Service
@Slf4j
public class WorkflowMonitoringService {
private final WorkflowInstanceRepository instanceRepository;
private final MeterRegistry meterRegistry;
private final Map<WorkflowState, Gauge> stateGauges = new ConcurrentHashMap<>();
public WorkflowMonitoringService(WorkflowInstanceRepository instanceRepository,
MeterRegistry meterRegistry) {
this.instanceRepository = instanceRepository;
this.meterRegistry = meterRegistry;
initializeMetrics();
}
private void initializeMetrics() {
// Create gauges for each workflow state
for (WorkflowState state : WorkflowState.values()) {
Gauge gauge = Gauge.builder("workflow.instances.by.state")
.tag("state", state.name())
.description("Number of workflow instances in state")
.register(meterRegistry, this, 
monitor -> instanceRepository.countByCurrentState(state));
stateGauges.put(state, gauge);
}
// Timer for workflow duration
Timer.builder("workflow.execution.duration")
.description("Workflow execution duration")
.register(meterRegistry);
}
public WorkflowMetrics getWorkflowMetrics() {
WorkflowMetrics metrics = new WorkflowMetrics();
metrics.setTotalInstances(instanceRepository.count());
metrics.setCompletedInstances(instanceRepository.countByCurrentState(WorkflowState.COMPLETED));
metrics.setFailedInstances(instanceRepository.countByCurrentState(WorkflowState.FAILED));
metrics.setActiveInstances(instanceRepository.countByCurrentStateNotIn(
List.of(WorkflowState.COMPLETED, WorkflowState.FAILED, WorkflowState.CANCELLED)));
// Calculate average duration
List<WorkflowInstance> completedInstances = instanceRepository
.findByCurrentState(WorkflowState.COMPLETED);
if (!completedInstances.isEmpty()) {
double avgDuration = completedInstances.stream()
.mapToLong(instance -> 
Duration.between(instance.getCreatedAt(), instance.getCompletedAt()).toMinutes())
.average()
.orElse(0.0);
metrics.setAverageDurationMinutes(avgDuration);
}
return metrics;
}
}
// Metrics DTO
public class WorkflowMetrics {
private long totalInstances;
private long completedInstances;
private long failedInstances;
private long activeInstances;
private double averageDurationMinutes;
// getters, setters
}

6. Compensation and Error Handling

Example 6: Advanced Compensation Handler

@Component
@Slf4j
public class CompensationHandler {
private final Map<String, CompensationStrategy> compensationStrategies;
public CompensationHandler(List<CompensationStrategy> strategies) {
this.compensationStrategies = strategies.stream()
.collect(Collectors.toMap(CompensationStrategy::getStepName, Function.identity()));
}
public CompensationResult executeCompensation(String workflowId, 
String failedStep,
Map<String, Object> context) {
log.info("Executing compensation for workflow: {}, failed step: {}", workflowId, failedStep);
List<CompensationStep> compensationSteps = determineCompensationSteps(failedStep);
CompensationResult result = new CompensationResult();
for (CompensationStep step : compensationSteps) {
try {
CompensationStrategy strategy = compensationStrategies.get(step.getStepName());
if (strategy != null) {
CompensationStepResult stepResult = strategy.compensate(context);
result.addStepResult(stepResult);
if (!stepResult.isSuccess()) {
log.error("Compensation failed for step: {}", step.getStepName());
result.setOverallSuccess(false);
// Continue with other compensations despite failures
}
} else {
log.warn("No compensation strategy found for step: {}", step.getStepName());
}
} catch (Exception e) {
log.error("Unexpected error during compensation for step: {}", step.getStepName(), e);
result.setOverallSuccess(false);
}
}
log.info("Compensation completed for workflow: {}, overall success: {}", 
workflowId, result.isOverallSuccess());
return result;
}
private List<CompensationStep> determineCompensationSteps(String failedStep) {
// Define compensation order based on the failed step
// This could be configured externally
Map<String, List<String>> compensationOrder = Map.of(
"SHIPPING", List.of("inventoryReservation", "paymentProcessing"),
"INVENTORY_RESERVATION", List.of("paymentProcessing"),
"PAYMENT_PROCESSING", List.of() // No compensation needed for payment failure
);
return compensationOrder.getOrDefault(failedStep, List.of()).stream()
.map(stepName -> new CompensationStep(stepName, CompensationPriority.HIGH))
.collect(Collectors.toList());
}
}
// Compensation Strategy Interface
public interface CompensationStrategy {
String getStepName();
CompensationStepResult compensate(Map<String, Object> context);
}
@Component
@Slf4j
public class PaymentCompensationStrategy implements CompensationStrategy {
private final PaymentService paymentService;
public PaymentCompensationStrategy(PaymentService paymentService) {
this.paymentService = paymentService;
}
@Override
public String getStepName() {
return "paymentProcessing";
}
@Override
public CompensationStepResult compensate(Map<String, Object> context) {
try {
String transactionId = (String) context.get("paymentTransactionId");
if (transactionId != null) {
log.info("Refunding payment transaction: {}", transactionId);
boolean refunded = paymentService.refundPayment(transactionId);
if (refunded) {
return CompensationStepResult.success("Payment refunded successfully");
} else {
return CompensationStepResult.failure("Payment refund failed");
}
}
return CompensationStepResult.success("No payment to refund");
} catch (Exception e) {
log.error("Payment compensation failed", e);
return CompensationStepResult.failure("Payment compensation error: " + e.getMessage());
}
}
}

Testing Long-Running Workflows

Example 7: Workflow Testing

@ExtendWith(SpringExtension.class)
@SpringBootTest
@TestPropertySource(properties = {
"spring.datasource.url=jdbc:h2:mem:testdb",
"spring.jpa.hibernate.ddl-auto=create-drop"
})
class WorkflowEngineTest {
@Autowired
private WorkflowEngine workflowEngine;
@Autowired
private WorkflowInstanceRepository instanceRepository;
@MockBean
private PaymentService paymentService;
@MockBean
private InventoryService inventoryService;
@Test
@DisplayName("Should complete workflow successfully")
void shouldCompleteWorkflowSuccessfully() {
// Given
OrderWorkflowContext context = new OrderWorkflowContext();
context.setOrderId("order-123");
context.setCustomerId("customer-456");
context.setAmount(new BigDecimal("99.99"));
when(paymentService.processPayment(any())).thenReturn(
new PaymentResponse(true, "txn-123", "credit_card", null));
when(inventoryService.reserveItems(any())).thenReturn(true);
// When
WorkflowInstance instance = workflowEngine.startWorkflow("order-processing", context);
// Then - wait for completion
await().atMost(10, TimeUnit.SECONDS).until(() -> {
WorkflowInstance updated = instanceRepository.findById(instance.getId()).orElseThrow();
return updated.getCurrentState() == WorkflowState.COMPLETED;
});
WorkflowInstance completed = instanceRepository.findById(instance.getId()).orElseThrow();
assertThat(completed.getCurrentState()).isEqualTo(WorkflowState.COMPLETED);
}
@Test
@DisplayName("Should handle payment failure with compensation")
void shouldHandlePaymentFailure() {
// Given
OrderWorkflowContext context = new OrderWorkflowContext();
context.setOrderId("order-456");
context.setCustomerId("customer-789");
context.setAmount(new BigDecimal("149.99"));
when(paymentService.processPayment(any())).thenReturn(
new PaymentResponse(false, null, null, "Insufficient funds"));
// When
WorkflowInstance instance = workflowEngine.startWorkflow("order-processing", context);
// Then - wait for failure and compensation
await().atMost(10, TimeUnit.SECONDS).until(() -> {
WorkflowInstance updated = instanceRepository.findById(instance.getId()).orElseThrow();
return updated.getCurrentState() == WorkflowState.FAILED;
});
WorkflowInstance failed = instanceRepository.findById(instance.getId()).orElseThrow();
assertThat(failed.getCurrentState()).isEqualTo(WorkflowState.FAILED);
assertThat(failed.getLastError()).contains("Payment failed");
}
}

Best Practices

1. Idempotency

@Component
@Slf4j
public class IdempotentWorkflowStep implements WorkflowStep {
private final WorkflowExecutionTracker executionTracker;
@Override
public WorkflowExecutionResult execute(WorkflowInstance instance, OrderWorkflowContext context) {
String executionKey = instance.getId() + ":" + getName();
// Check if already executed
if (executionTracker.isAlreadyExecuted(executionKey)) {
log.info("Step already executed: {}", executionKey);
return WorkflowExecutionResult.success("Step already executed");
}
// Execute step
WorkflowExecutionResult result = doExecute(instance, context);
if (result.isSuccess()) {
executionTracker.recordExecution(executionKey);
}
return result;
}
private WorkflowExecutionResult doExecute(WorkflowInstance instance, OrderWorkflowContext context) {
// Actual step implementation
return WorkflowExecutionResult.success("Step executed");
}
}

2. Configuration Externalization

# application.yml
workflow:
order-processing:
steps:
- name: VALIDATION
retry:
max-attempts: 3
backoff: exponential
initial-interval: 5000
- name: PAYMENT
retry:
max-attempts: 5
backoff: fixed
interval: 10000
- name: INVENTORY
retry:
max-attempts: 3
backoff: exponential
initial-interval: 5000
compensation:
order:
- PAYMENT
- INVENTORY

Conclusion

Long-running workflows in Java require careful consideration of:

Key Patterns:

  • State Machines: For complex state transitions
  • Saga Pattern: For distributed transactions
  • Compensation: For rollback logic
  • Persistence: For durability across restarts

Implementation Choices:

  • Custom Engine: Full control but more complexity
  • Temporal.io: Production-ready but external dependency
  • Spring State Machine: Spring integration but limited features
  • Database-driven: Simple but requires manual recovery

Critical Success Factors:

  1. Idempotency: Ensure steps can be safely retried
  2. Monitoring: Track workflow progress and performance
  3. Error Handling: Comprehensive failure and compensation strategies
  4. Testing: Thorough testing of all workflow paths
  5. Operational Support: Tools for manual intervention and debugging

Choose the approach that best fits your complexity requirements, team expertise, and operational capabilities.

Category : JAVA

Leave a Reply

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


Macro Nepal Helper