Chapter 9. Saga EIP

The In-memory Saga service is not recommended for production environments as it does not support persistence of the Saga status (it is kept only in-memory), so it cannot guarantee consistency of the Saga EIPs in case of application failure (for example, JVM crash). Also, when using a in-memory Saga service, Saga contexts cannot be propagated to remote services using transport-level headers (it can be done with other implementations). You can add the following code to customize the Camel context when you want to use the in-memory saga service. The service belongs to the camel-core module.

context.addService(new org.apache.camel.impl.saga.InMemorySagaService());

Some type of processing is required when the Saga is completed. Compensation endpoints are invoked when something wrong happens and the Saga is cancelled. The completion endpoints can be invoked to do further processing when the Saga is completed successfully. For example, in the order service above, we may need to know when the order is completed (and the credit reserved) to actually start preparing the order. We do not want to start to prepare the order if the payment is not done (unlike most modern CPUs that give you access to reserved memory before ensuring that you have rights to read it). This can be done easily with a modified version of the direct:newOrder endpoint:

from("direct:newOrder")
  .saga()
  .propagation(SagaPropagation.MANDATORY)
  .compensation("direct:cancelOrder")
  .completion("direct:completeOrder")
    .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
    .bean(orderManagerService, "newOrder")
    .log("Order ${body} created");

from("direct:completeOrder")
  .transform().header(Exchange.SAGA_LONG_RUNNING_ACTION)
  .bean(orderManagerService, "findExternalId")
  .to("jms:prepareOrder")
  .log("Order ${body} sent for preparation");

When the Saga is completed, the order is sent to a JMS queue for preparation. Like compensating actions, also completion actions may be called multiple times by the Saga coordinator (especially in case of errors, like network errors). In this example, the service listening to the prepareOrder JMS queue is prepared to hold possible duplicates (see the Idempotent Consumer EIP for examples on how to handle duplicates).

You can use Saga options to register custom identifiers. For example, the credit service is refactored as follows:

from("direct:reserveCredit")
  .bean(idService, "generateCustomId")
  .to("direct:creditReservation")

from("direct:creditReservation")
  .saga()
  .propagation(SagaPropagation.SUPPORTS)
  .option("CreditId", body())
  .compensation("direct:creditRefund")
    .bean(creditService, "reserveCredit")
    .log("Credit ${header.amount} reserved. Custom Id used is ${body}");

from("direct:creditRefund")
  .transform(header("CreditId")) // retrieve the CreditId option from headers
  .bean(creditService, "refundCredit")
  .log("Credit for Custom Id ${body} refunded");

The direct:creditReservation endpoint can be called outside of the Saga, by setting the propagation mode to SUPPORTS. This way multiple options can be declared in a Saga route.

Setting timeouts on Saga EIPs guarantees that a Saga does not remain stuck forever in the case of machine failure. The Saga EIP implementation has a default timeout set on all Saga EIPs that do not specify it explicitly. When the timeout expires, the Saga EIP will decide to cancel the Saga (and compensate all participants), unless a different decision has been taken before.

Timeouts can be set on Saga participants as follows:

from("direct:newOrder")
  .saga()
  .timeout(1, TimeUnit.MINUTES) // newOrder requires that the saga is completed within 1 minute
  .propagation(SagaPropagation.MANDATORY)
  .compensation("direct:cancelOrder")
  .completion("direct:completeOrder")
    // ...
    .log("Order ${body} created");

All participants (for example, credit service, order service) can set their own timeout. The minimum value of those timeouts is taken as timeout for the saga when they are composed together. A timeout can also be specified at the Saga level as follows:

from("direct:buy")
  .saga()
  .timeout(5, TimeUnit.MINUTES) // timeout at saga level
    .to("direct:newOrder")
    .to("direct:reserveCredit");

In the examples above, we have used the MANDATORY and SUPPORTS propagation modes, but also the REQUIRED propagation mode, that is the default propagation used when nothing else is specified. These propagation modes map 1:1 the equivalent modes used in transactional contexts.

When a Saga cannot be all executed in a synchronous way, but it requires, for example, communication with external services using asynchronous communication channels, then the completion mode cannot be set to AUTO (default), because the Saga is not completed when the exchange that creates it is done. This is often the case for the Saga EIPs that have long execution times (hours, days). In these cases, the MANUAL completion mode should be used.

from("direct:mysaga")
  .saga()
  .completionMode(SagaCompletionMode.MANUAL)
  .completion("direct:finalize")
  .timeout(2, TimeUnit.HOURS)
    .to("seda:newOrder")
    .to("seda:reserveCredit");

Add the asynchronous processing for seda:newOrder and seda:reserveCredit. These send the asynchronous callbacks to seda:operationCompleted.

from("seda:operationCompleted") // an asynchronous callback
  .saga()
  .propagation(SagaPropagation.MANDATORY)
    .bean(controlService, "actionExecuted")
    .choice()
      .when(body().isEqualTo("ok"))
        .to("saga:complete") // complete the current saga manually (saga component)
    .end()

You can add the direct:finalize endpoint to execute final actions.

Setting the completion mode to MANUAL means that the Saga is not completed when the exchange is processed in the route direct:mysaga but it will last longer (max duration is set to 2 hours). When both asynchronous actions are completed the Saga is completed. The call to complete is done using the Camel Saga Component’s saga:complete endpoint. There is a similar endpoint for manually compensating the Saga (saga:compensate).