Skip to content
OccurrentOccurrent

@Saga

Spring Boot

On the Spring Boot starter you do not wire a SagaRunner yourself. Annotate a no-arg factory method returning a Saga with org.occurrent.annotation.Saga. The framework registers it as a managed saga, subscribing through the same catch-up, durable-resume, and competing-consumer machinery as @Subscription:

import org.occurrent.annotation.Saga;


@Component
class OrderFulfillmentSaga {


    @Saga(id = "order-fulfillment")
    org.occurrent.dsl.saga.Saga<OrderEvent, FlowState<OrderEvent>, OrderCommand> orderFulfillment() {
        return FlowSaga.<OrderEvent, OrderCommand>builder()
                .correlateAll(OrderEvent::orderId)
                .startsOn(OrderPlaced.class,
                        order -> List.of(new ReservePayment(order.orderId(), order.amount())))
                .step("awaiting-payment", step -> step
                        .on(PaymentReserved.class, Continuation.end(),
                                payment -> List.of(new ShipOrder(payment.orderId())))
                        .on(PaymentFailed.class, Continuation.end(),
                                failure -> List.of(new CancelOrder(failure.orderId(), failure.reason())))
                        .timeout(Duration.ofMinutes(30), Continuation.end(),
                                received -> List.of(new CancelOrder(received.initiating(OrderPlaced.class).orderId(), "payment timeout"))))
                .build();
    }
}

The annotation and the DSL class share the name Saga, so a Java factory method has to qualify one of them, as above. Kotlin has no such clash, since the DSL entry point is the lowercase saga function and the return type can be inferred.

@Saga takes:

AttributeDescription
idThe durable subscription and checkpoint key (required).
startAtStartPosition.BEGINNING, NOW, or DEFAULT, the same start-position idea as @Subscription.
startAtGlobalPositionStart after a specific global position instead, to rewind a saga to a known point. Mutually exclusive with a non-default startAt.
resumeBehaviorDEFAULT or SAME_AS_START_AT, the same resume-behavior idea as @Subscription.
startupModeDEFAULT, WAIT_UNTIL_STARTED, or BACKGROUND, the same startup-mode idea as @Subscription. The default is BACKGROUND, so a saga replaying its history does not hold up application startup. Before 0.32.0 this attribute was accepted and ignored, and a replaying saga blocked startup, so set WAIT_UNTIL_STARTED if you depended on that. Timers wait for the replay to finish either way.
capabilityAGNOSTIC (both stream and DCB events) or STREAM (stream events only).
store / storeNameSelect the SagaStateStore bean by type or name. With both unset the store resolves by convention, the unique SagaStateStore bean, otherwise a zero-config MongoDB store in a saga-<id> collection.
commandDispatcher / commandDispatcherNameSelect the CommandDispatcher bean by type or name, otherwise the unique CommandDispatcher bean. There is no default dispatcher, since it is usually a lambda over your ApplicationService.
sourceEVENT_STORE (the default) reads the event store. PUSH feeds the saga from a PushSubscriptionModel bean instead, see Fed from a broker.
catchupFor a push saga only. FROM_EVENT_STORE (the default) replays history once before going live, NONE takes live events only and needs no event store.
subscriptionModel / subscriptionModelNameSelect the PushSubscriptionModel bean by type or name when source = PUSH.
redeliveryDetectionREQUIRED (the default) refuses an event the saga cannot recognize a redelivery of. BEST_EFFORT takes it anyway with one warning, see Forward the Occurrent extensions.

@Saga is blocking-only in this first version, the reactive starter does not register it.

Fed from a broker (Added in 0.32.0)

Section titled “Fed from a broker”

A saga does not have to read the event store. Set source = Source.PUSH and point it at a PushSubscriptionModel bean, and it reacts to whatever your listener hands that model, from RabbitMQ, Kafka, or any other source that can deliver an event again:

import org.occurrent.annotation.Saga;
import org.occurrent.annotation.Source;


@Component
class OrderFulfillmentSaga {


    @Saga(id = "order-fulfillment", source = Source.PUSH, subscriptionModelName = "orderEvents")
    org.occurrent.dsl.saga.Saga<OrderEvent, FlowState<OrderEvent>, OrderCommand> orderFulfillment() {
        return FlowSaga.<OrderEvent, OrderCommand>builder()
                .correlateAll(OrderEvent::orderId)
                .startsOn(OrderPlaced.class,
                        order -> List.of(new ReservePayment(order.orderId(), order.amount())))
                .step("awaiting-payment", step -> step
                        .on(PaymentReserved.class, Continuation.end(),
                                payment -> List.of(new ShipOrder(payment.orderId()))))
                .build();
    }
}


@Configuration
class OrderEventsConfig {


    @Bean("orderEvents")
    PushSubscriptionModel orderEvents() {
        return new PushSubscriptionModel();
    }
}

Your listener calls acceptOrDefer(cloudEvent) on that bean and acts on the RoutingOutcome it returns, as shown for the PushSubscriptionModel. The saga takes it from there, with the same correlateAll, steps, timeouts and state store it always used.

An event the saga does not declare is ignored rather than refused. The saga subscribes with a filter on its declared event types. So when the queue delivers something else, acceptOrDefer returns FILTERED and your listener acknowledges as usual, with nothing logged. A queue that also delivers event types the saga never declared therefore works fine, and each extra event costs one filter check.

The redelivery check described under Forward the Occurrent extensions below only runs for events the saga declares. So a listener that drops the Occurrent extensions shows up only once a declared event arrives without them.

subscriptionModelName is only needed when there is something to choose between. A PushSubscriptionModel feeds exactly one saga or projection, so an application with several push sagas declares several feed beans, and each @Saga names its own. With a single feed bean, as here, the name can be dropped and the starter finds the bean on its own. It refuses to guess between several, with a message naming them.

Declaring the bean also does not replace the default subscription model the starter contributes. The starter skips models without a start position, checkpoint or catch-up when it decides whether the application brought its own subscription model, and a push feed has none of the three, so your event-store sagas and subscriptions keep the durable default and the feed runs beside it.

By default the starter puts a replay in front of the feed, so a saga that has never run works through the event store’s history first and only then starts taking live events. That is what you want when this application wrote the events and the broker is only how they reach the saga.

When the events are not in your event store

Section titled “When the events are not in your event store”

If another application writes the events, your event store does not hold them, so there is nothing to replay. A replay would either find nothing or, worse, apply unrelated events that happen to live in the same store. Say so with catchup = Catchup.NONE:

@Saga(id = "shipment-tracking", source = Source.PUSH, subscriptionModelName = "warehouseEvents",
      catchup = Catchup.NONE)
org.occurrent.dsl.saga.Saga<ShipmentEvent, FlowState<ShipmentEvent>, ShipmentCommand> shipmentTracking() {
    return /* ... */;
}

This takes live events only and touches no event store at all, so the application needs neither a PositionOrderedReader nor a CheckpointStorage bean. The bean requirements and failure messages are the same as for a push projection with catchup = Catchup.NONE. Leave the default in place without those beans and startup fails with a message telling you to set catchup = Catchup.NONE, rather than a bare missing-bean error.

A saga that has run before picks up where it left off either way, because its per-instance state lives in its SagaStateStore and not in the feed. The difference shows on a first run against an existing history. With the default the saga is brought up to date from the event store before it reacts. With Catchup.NONE it starts from nothing and reacts only to what arrives from here on.

Forward the Occurrent extensions

Section titled “Forward the Occurrent extensions”

A saga recognizes a redelivered event by its streamid together with its streamversion, or by its position. That pair, or the position, is the event’s redelivery key. A broker delivers at least once, so an event with none of those would make the saga react a second time and issue its commands again on every redelivery.

The saga therefore refuses such an event by throwing SagaRedeliveryDetectionException before the reaction runs, instead of acknowledging it. What happens to the event after that is up to your feed. Whatever the feed does, the listener that dropped the metadata sees the exception, and that listener is where it can be fixed.

Occurrent’s own stored events always have the extensions, so this only happens with events your listener forwards without them.

Your feed might carry none of that redelivery metadata, another application’s broker for example, while every command the saga issues is still safe to receive more than once. When both are true, opt out with @Saga(redeliveryDetection = RedeliveryDetection.BEST_EFFORT), or SagaRunnerConfig.withRedeliveryDetection(BEST_EFFORT) when you drive SagaRunner yourself. The saga then takes those events and logs one warning, naming the saga so you can find it. Setting BEST_EFFORT on an event-store saga (source = EVENT_STORE) is rejected at startup instead, since those events always carry the extensions and there is no metadata gap for it to change. The reasoning is in ADR 0109.

This is also why a @Saga accepts only a PushSubscriptionModel and not a DomainEventFeed, which a @Projection does accept. A domain event feed carries no stream metadata, so a saga bound to one would lose its redelivery protection without saying anything.

What a push saga cannot set

Section titled “What a push saga cannot set”

startAt, startAtGlobalPosition and resumeBehavior are rejected rather than quietly ignored. A replay always starts at the beginning, and where the live feed resumes after a restart is the broker’s business. Occurrent only records that the catch-up finished, so a restart skips the replay and the broker redelivers whatever the consumer had not acknowledged.

startupMode does apply, under the default catchup, because that replay is real work on the startup path. Use StartupMode.BACKGROUND to keep a long history from holding up startup. It is rejected together with Catchup.NONE, where there is no replay to wait for. Setting catchup on an event-store saga is rejected too, since ignoring it would leave the saga reading the whole history it was asked to skip.

Starting it yourself

Section titled “Starting it yourself”

A push feed is a bean you supply, so occurrent.subscription.mode=manual cannot withhold it the way it withholds a subscription Occurrent owns. Inject ManualStartPushSources and start the saga when the application is ready, which is what you want for a saga behind a leader election, where every node would otherwise react to the same event:

@Autowired
ManualStartPushSources pushSources;


void becameLeader() {
    pushSources.start("order-fulfillment");
}

startAll() starts every withheld push source, projections and sagas alike, and pendingIds() says what is still waiting. The saga’s instances are readable through SagaInstances while it waits, since that reads the state store rather than a running subscription, so you can look at what is in flight before deciding to start it.

The design rationale, including why a domain event feed is refused and why the timers wait for the handover, is in ADR 0096.

Observing @Saga instances

Section titled “Observing @Saga instances”

The @Saga registrar publishes each saga’s SagaInstances under a registry keyed by saga id:

@Service
class SagaDashboard {


    private final SagaInstancesRegistry registry;


    SagaDashboard(SagaInstancesRegistry registry) {
        this.registry = registry;
    }


    List<SagaInstance> stalled(String sagaId, Duration threshold) {
        SagaInstances instances = registry.get(sagaId);
        List<SagaInstance> stalled = new ArrayList<>(
                instances.findByStatus(SagaStatus.ACTIVE, Instant.now().minus(threshold), 100));
        stalled.addAll(instances.findByStatus(SagaStatus.QUARANTINED, Instant.now(), 100));
        return stalled;
    }
}

get(id) throws and names every id that is registered, which is what you want when the id is a constant in your own code. find(id) returns an Optional for an id that came from a request or a configuration value. sagaIds() lists them so a dashboard does not hardcode ids. Each saga is also published under the bean name sagaInstances-<id>, reachable with getBean or a @Qualifier if you prefer to inject one saga’s view directly.

Each saga’s running SagaSubscription is published the same way, as sagaSubscription-<id>. It’s the handle SagaRunner returns on the programmatic path, so it gives you the saga’s event Subscription and waitUntilStarted() as well as instances(). A saga on a push feed that you start through ManualStartPushSources gets this bean when you start it, not when the context refreshes.

One timing constraint comes with the annotation path. A @Saga factory can only run once the beans it collaborates with are wired, which is after the context has refreshed, so the registry holds nothing until that scan has run. Inject it and read it when a request arrives, never from another bean’s constructor.

You never build a SagaRunnerConfig for a @Saga, so the quarantine budget is the property occurrent.saga.quarantine-after, one value for every @Saga in the application. It has no default. Leave it out, or leave it blank, to keep quarantine off, or set it to a positive duration to turn it on. Zero or a negative value stops an application with at least one @Saga from starting.

occurrent.saga.quarantine-after=5m

Several instances of a @Saga

Section titled “Several instances of a @Saga”

The starter uses a competing-consumer subscription model by default, so only one instance receives a saga’s events at a time. Gating the timer poller is on by default too. A @Saga runner reuses the same SpringMongoLeaseCompetingConsumerStrategy the subscription model already builds, so nothing extra to wire. Both the poll interval and the gating are configurable under occurrent.saga:

occurrent:
  saga:
    timer-poll-interval: 15s   # the default, lower it only if you rely on short timeouts firing promptly
    competing-consumer:
      enabled: true   # set to false to let every instance poll (the uncoordinated behavior)

Turn competing-consumer.enabled off and the poller runs on every instance.