SagaTemplate and Event Listeners

SagaTemplate

SagaTemplate is the main entry point for interacting with the StackSaga engine (SEC) in an orchestrator service. It provides a fluent builder API to initialize transactions, configure execution modes (fire-and-forget or fire-and-watch), and query the current state and execution history of any transaction.

SagaTemplate seamlessly supports both imperative (blocking) and reactive (non-blocking) application architectures.

SagaTemplate in a Non-Reactive Environment

In a traditional Spring MVC or blocking environment, use SagaTemplate to initiate saga transactions and retrieve transaction snapshots synchronously:

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler {

    private final SagaTemplate<OrderDomainEntity> sagaTemplate; (1)

    public String handle(String username, double amount, List<String> items) {
        String transactionId = this.sagaTemplate
                (2)
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    {//initializing the domain entity with the required properties for the saga execution.
                        orderDomainEntity.setUsername(username);
                        orderDomainEntity.setTotalAmount(amount);
                        //...
                    }
                    return orderDomainEntity;
                })
                (3)
                .peek(orderDomainEntity -> {
                    log.info("transactionId {}:", orderDomainEntity.getTransactionId());
                })
                (4)
                .startWith(ValidateUserExecutor.class)
                (5)
                .fireAndForget()
                (6)
                .execute();
        log.info("Started place order saga with transactionId: {}", transactionId);
        return transactionId;
    }

    public void printCurrentState(String transactionId) {
        TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> state = this
                .sagaTemplate
                (7)
                .getCurrentState(transactionId)
                .fetch();

        TransactionCompleteStatus currentStatus = state.getCurrentStatus(); (8)
        log.info("Current status of transaction {}: {}", transactionId, currentStatus);
        NavigableMap<Integer, ? extends SyncExecutionEvent<OrderDomainEntity>> executionHistory = state.getExecutionHistory(); (9)
        log.info("Execution history of transaction {}: {}", transactionId, executionHistory);
        OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity(); (10)
        log.info("Current domain entity state: {}", currentDomainEntity); (11)
        ZonedDateTime startedDateTime = state.getStartedDateTime(); (12)
        log.info("Transaction {} started at: {}", transactionId, startedDateTime.toLocalDateTime());
    }
}
1 Inject SagaTemplate configured for your custom DomainEntity type.
2 Use init() to provide a Supplier creating the initial DomainEntity. The framework generates and assigns the unique transactionId during initialization.
3 Use peek() (optional) to inspect or enrich the initialized DomainEntity before execution begins.
4 Use startWith() to specify the initial executor that starts the transaction flow.
5 Use fireAndForget() to select the fire-and-forget execution mode.
6 Call execute() to initiate the transaction in a blocking manner. It returns the generated transactionId once initialized.
7 Call getCurrentState(transactionId).fetch() to retrieve a synchronous snapshot of the transaction state from the event-store.
8 Retrieve the overall completion status (TransactionCompleteStatus) of the transaction.
9 Retrieve the chronological execution history (NavigableMap<Integer, ? extends SyncExecutionEvent<DE>>) containing every executed span (primary and compensation) with metadata and status.
10 Retrieve the current domain entity snapshot representing all state mutations applied up to this point.
11 Log the domain entity snapshot.
12 Retrieve the start timestamp of the transaction.
For real-time updates without polling, implement a State Change Listener.

SagaTemplate in a Reactive Environment

In a reactive environment (e.g., Spring WebFlux), SagaTemplate provides non-blocking operators across all lifecycle phases.

There are two primary modes to execute a saga transaction reactively:

  1. Fire-and-Forget — Start the transaction and immediately receive a Mono<String> emitting the transaction ID once the saga is accepted. The saga executes in the background. Use this when you only need the transaction ID and will rely on getCurrentState(transactionId).fetchAsync() or the EventListener to track progress.

  2. Fire-and-Watch — Start the transaction and observe real-time state transitions as a Flux<TransactionState<DE, SyncExecutionEvent<DE>>>. Ideal for streaming live execution events to clients (e.g., via Server-Sent Events or WebSockets).

Fire-and-Forget

Calling .fireAndForget().executeAsync() hands the transaction off to the StackSaga engine and returns a Mono<String> emitting the generated transaction ID:

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler {

    private final SagaTemplate<OrderDomainEntity> sagaTemplate;

    public Mono<String> handle(String username, double amount, List<String> items) {
        return this.sagaTemplate
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    {//initializing the domain entity with the required properties for the saga execution.
                        orderDomainEntity.setUsername(username);
                        orderDomainEntity.setTotalAmount(amount);
                        //...
                    }
                    return orderDomainEntity;
                })
                .peek(orderDomainEntity -> {
                    log.info("transactionId {}:", orderDomainEntity.getTransactionId());
                })
                .startWith(ValidateUserExecutor.class)
                .fireAndForget()
                (1)
                .executeAsync()
                .map(transactionId -> "Order placed successfully with transaction id: " + transactionId);
    }
}
1 executeAsync() returns a Mono<String> that emits the transaction ID once the saga is initialised. The transaction continues running in the background; use getCurrentState(transactionId).fetchAsync() or the EventListener to track its progress.

Fire-and-Watch

Calling .fireAndWatch(Duration waitTimeout).toFlux() initiates the transaction and returns a Flux<TransactionState<DE, SyncExecutionEvent<DE>>> that emits an updated TransactionState snapshot after each executor span completes:

@GetMapping(value = "/order/watch", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>>> createOrderAndWatch() {
    return sagaTemplate
            .init(OrderDomainEntity::new)
            .peek(orderDomainEntity -> {
                log.info("Started Transaction ID: {}", orderDomainEntity.getTransactionId());
            })
            .startWith(ValidateUserExecutor.class)
            .fireAndWatch(Duration.ofSeconds(40)) (1)
            .toFlux()
            .doOnNext(state -> {
                log.info("Received state change: {}", state);
            });
}
1 waitTimeout defines the maximum duration the subscriber will wait for execution results before the pipeline signals a WaitTimeExceededException.
Key Behaviors and Error Handling

  • Timeout (WaitTimeExceededException): If execution exceeds the specified Duration waitTimeout, the pipeline terminates with a WaitTimeExceededException via onError. This only ends the subscription — the saga execution continues asynchronously in the background. After the subscription ends, you can still observe progress in two ways:

    • If you have registered a State Change Listener (TransactionEventListener or ReactiveTransactionEventListener), it continues to receive event callbacks.

    • Alternatively, query the state at any time using getCurrentState(transactionId).fetchAsync().

  • Execution Pause (TransactionPausedException): If a step encounters a retryable exception (such as a transient database failure or custom RetryableExecutorException), execution is paused and rescheduled. Because TransactionState only represents completed saga states, the live subscription halts and emits a TransactionPausedException via onError. This indicates a temporary retry pause, not a fatal failure. Once rescheduled and retried, state notifications continue via listeners, or you can query state using getCurrentState(transactionId).fetchAsync().

Observing Transaction State Changes

After initiating a transaction via execute() or executeAsync(), the StackSaga engine executes each step sequentially (forward commands/queries and backward compensations).

There are two distinct mechanisms to observe state changes:

  1. State Change Listeners — Push-based: The engine notifies your listener callback automatically after each executor step completes. Best for real-time reactions such as dispatching notifications, updating caches, or emitting events.

  2. SagaTemplate.getCurrentState() — Pull-based: Queries the event-store on demand to retrieve the latest state snapshot for a given transaction ID. Best for user-facing status queries and post-completion audits.

Using State Change Listeners (Real-time)

State Change Listeners receive push notifications directly from the engine after every executor span completes (both forward execution and compensations).

Choose the listener interface matching your application architecture:

  1. TransactionEventListener — For imperative / blocking environments.

  2. ReactiveTransactionEventListener — For reactive / non-blocking environments.

Sequential Execution and Latency Impact:

The onStateChanged callback is executed sequentially as part of the transaction lifecycle. Specifically, the next span/executor step in the saga flow is triggered only after the current onStateChanged execution has completed.

Performing heavy computations, slow I/O calls, or long-running operations inside onStateChanged directly blocks the progression of the transaction and increases total transaction latency.

Best Practice: Keep onStateChanged lightweight (e.g., publishing a quick notification or logging status). If you need to perform heavy or time-consuming operations, offload the work to an asynchronous thread pool (using a snapshot or deep copy of TransactionState if needed), allowing the engine to immediately proceed with the next span of the transaction.

TransactionEventListener

Implement TransactionEventListener to listen to transaction events in an imperative environment:

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler implements TransactionEventListener<OrderDomainEntity> {

    @Override
    public void onStateChanged(TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> state) {
        TransactionCompleteStatus currentStatus = state.getCurrentStatus();
        log.info("Current status of transaction: {}", currentStatus);
        NavigableMap<Integer, ? extends SyncExecutionEvent<OrderDomainEntity>> executionHistory = state.getExecutionHistory();
        log.info("Execution history of transaction: {}", executionHistory);
        OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity();
        log.info("Current domain entity state: {}", currentDomainEntity);
        ZonedDateTime startedDateTime = state.getStartedDateTime();
        log.info("Transaction started at: {}", startedDateTime.toLocalDateTime());
    }
}
Because onStateChanged is synchronous, blocking user code, the engine invokes it on a dedicated bounded-elastic scheduler (blockingExecutionScheduler) supplied by AbstractSchedulerProvider. This isolates listener execution from the engine’s core reactive threads. The thread cap, queue capacity, and TTL can be tuned via the stacksaga.scheduler.blocking-execution.* configuration properties. See Configuration Properties.

ReactiveTransactionEventListener

Implement ReactiveTransactionEventListener to listen to transaction events in a reactive environment:

@Slf4j
@Component
@RequiredArgsConstructor
public class ReactivePlaceOrderHandler implements ReactiveTransactionEventListener<OrderDomainEntity> {

    @Override
    public Mono<Void> onStateChanged(TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> transactionState) {
        return Mono
                .just(transactionState)
                .doOnNext(state -> {
                    TransactionCompleteStatus currentStatus = state.getCurrentStatus();
                    log.info("Current status of transaction: {}", currentStatus);
                    NavigableMap<Integer, ? extends SyncExecutionEvent<OrderDomainEntity>> executionHistory = state.getExecutionHistory();
                    log.info("Execution history of transaction: {}", executionHistory);
                    OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity();
                    log.info("Current domain entity state: {}", currentDomainEntity);
                    ZonedDateTime startedDateTime = state.getStartedDateTime();
                    log.info("Transaction started at: {}", startedDateTime.toLocalDateTime());
                })
                .flatMap(state -> {
                    // simulate async non-blocking processing
                    return Mono.delay(Duration.ofSeconds(1)).then();
                });
    }
}
Unlike TransactionEventListener, ReactiveTransactionEventListener#onStateChanged runs on the shared non-blocking executionScheduler. The returned Mono<Void> must be fully non-blocking; any blocking calls will starve worker threads and affect other concurrent transactions. Parallelism can be tuned via stacksaga.scheduler.execution.parallelism. See Configuration Properties.

Using SagaTemplate.getCurrentState() (On-demand)

SagaTemplate.getCurrentState(String transactionId) provides pull-based access to the complete state snapshot and execution history of any transaction by its ID.

Calling getCurrentState(transactionId) returns a SagaTemplate.StateSpec<DE> specification that provides two methods:

  • fetch() — Synchronous retrieval returning TransactionState<DE, SyncExecutionEvent<DE>>.

    Do not call fetch() on non-blocking Reactor or Netty event loop threads. The method performs a thread check (Schedulers.isInNonBlockingThread()) and throws an IllegalStateException if invoked from a non-blocking thread. In reactive applications, always use fetchAsync().

  • fetchAsync() — Asynchronous retrieval returning Mono<TransactionState<DE, SyncExecutionEvent<DE>>>. Safe for use in Spring WebFlux and reactive event-loop threads.

Blocking State Retrieval (Imperative / Spring MVC)

TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> state =
        this.sagaTemplate
                .getCurrentState(transactionId)
                .fetch();

TransactionCompleteStatus status = state.getCurrentStatus();
NavigableMap<Integer, ? extends SyncExecutionEvent<OrderDomainEntity>> history = state.getExecutionHistory();
OrderDomainEntity entity = state.getCurrentDomainEntity();

Reactive State Retrieval (Reactive / Spring WebFlux)

Mono<TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>>> stateMono =
        this.sagaTemplate
                .getCurrentState(transactionId)
                .fetchAsync();

stateMono.subscribe(state -> {
    log.info("Transaction status: {}", state.getCurrentStatus());
});
Unlike State Change Listeners which receive pushed state events in memory as execution proceeds, getCurrentState(transactionId) queries the underlying event-store on every call. It is designed for on-demand queries, rather than high-frequency polling during active execution.

The StackSaga team recommends organizing saga transaction logic inside a dedicated Handler component that encapsulates transaction initiation, state queries, and lifecycle event observation.

Imperative Handler Pattern

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler implements TransactionEventListener<OrderDomainEntity> {

    private final SagaTemplate<OrderDomainEntity> sagaTemplate;

    public String handle(String username, double amount, List<String> items) { (1)
        return this.sagaTemplate
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    orderDomainEntity.setUsername(username);
                    orderDomainEntity.setTotalAmount(amount);
                    return orderDomainEntity;
                })
                .startWith(ValidateUserExecutor.class)
                .fireAndForget()
                .execute();
    }

    public CustomOrderStatusView getCurrentState(String orderId) { (2)
        TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> state = this
                .sagaTemplate
                .getCurrentState(orderId)
                .fetch();
        return new CustomOrderStatusView(state);
    }

    @Override
    public void onStateChanged(TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> transactionState) { (3)
        log.info("Transaction {} updated to status {}",
                transactionState.getTransactionId(),
                transactionState.getCurrentStatus());
    }
}
1 The handle method initiates the saga transaction using SagaTemplate and returns the generated transaction ID.
2 Retrieve the current transaction state on demand using .getCurrentState(orderId).fetch().
3 Implement onStateChanged from TransactionEventListener to handle real-time lifecycle notifications.

Reactive Handler Pattern

@Slf4j
@Component
@RequiredArgsConstructor
public class ReactivePlaceOrderHandler implements ReactiveTransactionEventListener<OrderDomainEntity> {

    private final SagaTemplate<OrderDomainEntity> sagaTemplate;

    public Mono<String> handle(String username, double amount, List<String> items) { (1)
        return this.sagaTemplate
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    orderDomainEntity.setUsername(username);
                    orderDomainEntity.setTotalAmount(amount);
                    return orderDomainEntity;
                })
                .startWith(ValidateUserExecutor.class)
                .fireAndForget()
                .executeAsync();
    }

    public Mono<CustomOrderStatusView> getCurrentState(String orderId) { (2)
        return this.sagaTemplate
                .getCurrentState(orderId)
                .fetchAsync()
                .map(CustomOrderStatusView::new);
    }

    @Override
    public Mono<Void> onStateChanged(TransactionState<OrderDomainEntity, SyncExecutionEvent<OrderDomainEntity>> transactionState) { (3)
        return Mono.fromRunnable(() ->
                log.info("Transaction {} updated to status {}",
                        transactionState.getTransactionId(),
                        transactionState.getCurrentStatus())
        );
    }
}
1 Non-blocking transaction initiation returning Mono<String> with the transaction ID.
2 Non-blocking state query using .getCurrentState(orderId).fetchAsync().
3 Implement onStateChanged from ReactiveTransactionEventListener returning Mono<Void> for non-blocking event handling.