Chapter 9. Writing a Camel application that uses transactions

Figure 9.1, “Demarcation by Marking the Route” shows an example of a route that is made transactional by adding the transacted() DSL command to the route. All of the route nodes that follow the transacted() node are included in the transaction scope. In this example, the two following nodes access a JDBC resource.

The following Java DSL example shows how to define a transactional route by marking the route with the transacted() DSL command:

import org.apache.camel.builder.RouteBuilder;

class MyRouteBuilder extends RouteBuilder {
    public void configure() {
        from("file:src/data?noop=true")
                .transacted()
                .bean("accountService","credit")
                .bean("accountService","debit");
    }
}

In this example, the file endpoint reads some XML format files that describe a transfer of funds from one account to another. The first bean() invocation credits the specified sum of money to the beneficiary’s account and then the second bean() invocation subtracts the specified sum of money from the sender’s account. Both of the bean() invocations cause updates to be made to a database resource. It is assumed that the database resource is bound to the transaction through the transaction manager, for example, see Chapter 6, Using JDBC data sources.

The preceding route can also be expressed in Blueprint XML. The <transacted /> tag marks the route as transactional, as shown in the following XML:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ...>

    <camelContext xmlns="http://camel.apache.org/schema/blueprint">
        <route>
            <from uri="file:src/data?noop=true" />
            <transacted />
            <bean ref="accountService" method="credit" />
            <bean ref="accountService" method="debit" />
        </route>
    </camelContext>

</blueprint>

To demarcate transactions, the transacted processor must be associated with a particular transaction manager instance. To save you having to specify the transaction manager every time you invoke transacted(), the transacted processor automatically picks a sensible default. For example, if there is only one instance of a transaction manager in your configuration, the transacted processor implicitly picks this transaction manager and uses it to demarcate transactions.

A transacted processor can also be configured with a transacted policy, of TransactedPolicy type, which encapsulates a propagation policy and a transaction manager (see Section 9.4, “Transaction propagation policies” for details). The following rules are used to pick the default transaction manager or transaction policy:

You also have the option of specifying a bean explicitly by providing the bean ID as an argument to transacted(). See Section 9.4.4, “Sample route with PROPAGATION_NEVER policy in Java DSL”.

If you insert a transacted processor into a route, the transaction manager creates a new transaction each time an exchange passes through this node. The transaction’s scope is defined as follows:

Any route nodes that precede the transacted processor are not in the transaction. However, if the route begins with a transactional endpoint then all nodes in the route are in the transaction. See Section 9.2.5, “Transactional endpoints at start of route”.

Consider the following route. It is incorrect because the transacted() DSL command mistakenly appears after the first bean() call, which accesses the database resource:

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
                .bean("accountService", "credit")
                .transacted()  // <-- WARNING: Transaction started in the wrong place!
                .bean("accountService", "debit");
    }
}

It is crucial to understand that a given transaction is associated with only the current thread. You must not create a thread pool in the middle of a transactional route because the processing in the new threads will not participate in the current transaction. For example, the following route is bound to cause problems:

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
                .transacted()
                .threads(3)  // WARNING: Subthreads are not in transaction scope!
                .bean("accountService", "credit")
                .bean("accountService", "debit");
    }
}

A route such as the preceding one is certain to corrupt your database because the threads() DSL command is incompatible with transacted routes. Even if the threads() call precedes the transacted() call, the route will not behave as expected.

If you want to break a route into fragments and have each route fragment participate in the current transaction, you can use direct: endpoints. For example, to send exchanges to separate route fragments, depending on whether the transfer amount is big (greater than 100) or small (less than or equal to 100), you can use the choice() DSL command and direct endpoints, as follows:

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
                .transacted()
                .bean("accountService", "credit")
                .choice().when(xpath("/transaction/transfer[amount > 100]"))
                .to("direct:txbig")
                .otherwise()
                .to("direct:txsmall");

        from("direct:txbig")
                .bean("accountService", "debit")
                .bean("accountService", "dumpTable")
                .to("file:target/messages/big");

        from("direct:txsmall")
                .bean("accountService", "debit")
                .bean("accountService", "dumpTable")
                .to("file:target/messages/small");
    }
}

Both the fragment beginning with direct:txbig and the fragment beginning with direct:txsmall participate in the current transaction because the direct endpoints are synchronous. This means that the fragments execute in the same thread as the first route fragment and, therefore, they are included in the same transaction scope.

The following Apache Camel components act as resource endpoints when they appear as the destination of a route, for example, if they appear in the to() DSL command. That is, these endpoints can access a transactional resource, such as a database or a persistent queue. The resource endpoints can participate in the current transaction, as long as they are associated with the same transaction manager as the transacted processor that initiated the current transaction.

The following example shows a route with resource endpoints. This sends the order for a money transfer to two different JMS queues. The credits queue processes the order to credit the receiver’s account. The debits queue processes the order to debit the sender’s account. There should be a credit only if there is a corresponding debit. Consequently, you want to enclose the enqueueing operations in a single transaction. If the transaction succeeds, both the credit order and the debit order will be enqueued. If an error occurs, neither order will be enqueued.

from("file:src/data?noop=true")
        .transacted()
        .to("jmstx:queue:credits")
        .to("jmstx:queue:debits");

Section 9.2, “Demarcation by transactional endpoints” shows an example of a route that is made transactional by the presence of a transactional endpoint at the start of the route (in the from() command). All of the route nodes are included in the transaction scope. In this example, all of the endpoints in the route access a JMS resource.

The following Java DSL example shows how to define a transactional route by starting the route with a transactional endpoint:

from("jmstx:queue:giro")
        .to("jmstx:queue:credits")
        .to("jmstx:queue:debits");

In the previous example, the transaction scope encompasses the endpoints, jmstx:queue:giro, jmstx:queue:credits, and jmstx:queue:debits. If the transaction succeeds, the exchange is permanently removed from the giro queue and pushed on to the credits queue and the debits queue. If the transaction fails, the exchange does not get put on to the credits and debits queues and the exchange is pushed back on to the giro queue. By default, JMS automatically attempts to redeliver the message. The JMS component bean, jmstx, must be explicitly configured to use transactions, as follows:

<blueprint ...>
    <bean id="jmstx" class="org.apache.camel.component.jms.JmsComponent">
        <property name="configuration" ref="jmsConfig" />
    </bean>

    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="jmsConnectionFactory" />
        <property name="transactionManager" ref="jmsTransactionManager" />
        <property name="transacted" value="true" />
    </bean>
    ...
</blueprint>

In the previous example, the transaction manager instance, jmsTransactionManager, is associated with the JMS component and the transacted property is set to true to enable transaction demarcation for InOnly exchanges.

The preceding route can equivalently be expressed in Blueprint XML, as follows:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0">

    <camelContext xmlns="http://camel.apache.org/schema/blueprint">
        <route>
            <from uri="jmstx:queue:giro" />
            <to uri="jmstx:queue:credits" />
            <to uri="jmstx:queue:debits" />
        </route>
    </camelContext>

</blueprint>

The transacted() DSL command is not required in a route that starts with a transactional endpoint. Nevertheless, assuming that the default transaction policy is PROPAGATION_REQUIRED (see Section 9.4, “Transaction propagation policies”), it is usually harmless to include the transacted() command, as in this example:

from("jmstx:queue:giro")
        .transacted()
        .to("jmstx:queue:credits")
        .to("jmstx:queue:debits");

However, it is possible for this route to behave in unexpected ways, for example, if a single TransactedPolicy bean having a non-default propagation policy is created in Blueprint XML. See Section 9.1.4, “Default transaction manager and transacted policy”. Consequently, it is usually better not to include the transacted() DSL command in routes that start with a transactional endpoint.

The following Apache Camel components act as transactional endpoints when they appear at the start of a route (for example, if they appear in the from() DSL command). That is, these endpoints can be configured to behave as a transactional client and they can also access a transactional resource.

To declare transaction policies at the bean level, insert a tx:transaction element as a child of the bean element, as follows:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
        xmlns:tx="http://aries.apache.org/xmlns/transactions/v1.1.0">

    <bean id="accountFoo" class="org.jboss.fuse.example.Account">
        <tx:transaction method="*" value="Required" />
        <property name="accountName" value="Foo" />
    </bean>

    <bean id="accountBar" class="org.jboss.fuse.example.Account">
        <tx:transaction method="*" value="Required" />
        <property name="accountName" value="Bar" />
    </bean>

</blueprint>

In the preceding example, the required transaction policy is applied to all methods of the accountFoo bean and the accountBar bean, where the method attribute specifies the wildcard, *, to match all bean methods.

To declare transaction policies at the top level, insert a tx:transaction element as a child of the blueprint element, as follows:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
        xmlns:tx="http://aries.apache.org/xmlns/transactions/v1.1.0">

    <tx:transaction bean="account*" value="Required" />

    <bean id="accountFoo" class="org.jboss.fuse.example.Account">
        <property name="accountName" value="Foo" />
    </bean>

    <bean id="accountBar" class="org.jboss.fuse.example.Account">
        <property name="accountName" value="Bar" />
    </bean>

</blueprint>

In the preceding example, the Required transaction policy is applied to all methods of every bean whose ID matches the pattern, account*.

The tx:transaction element supports the following attributes:

Apache Camel lets you define Spring transaction policies using the org.apache.camel.spring.spi.SpringTransactionPolicy class, which is essentially a wrapper around a native Spring class. The SpringTransactionPolicy class encapsulates two pieces of data:

For example, you could instantiate a Spring transaction policy bean with PROPAGATION_MANDATORY behavior, as follows:

<blueprint ...>
  <bean id="PROPAGATION_MANDATORY "class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    <property name="transactionManager" ref="txManager" />
    <property name="propagationBehaviorName" value="PROPAGATION_MANDATORY" />
  </bean>
  ...
</blueprint>

The following propagation behaviors are supported by Spring. These values were originally modeled on the propagation behaviors supported by JavaeEE:

The following example shows how to define transaction policy beans for all of the supported propagation behaviors. For convenience, each of the bean IDs matches the specified value of the propagation behavior value, but in practice you can use whatever value you like for the bean IDs.

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <bean id="PROPAGATION_MANDATORY " class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_MANDATORY" />
    </bean>

    <bean id="PROPAGATION_NESTED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_NESTED" />
    </bean>

    <bean id="PROPAGATION_NEVER" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_NEVER" />
    </bean>

    <bean id="PROPAGATION_NOT_SUPPORTED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_NOT_SUPPORTED" />
    </bean>

    <!-- This is the default behavior. -->
    <bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
    </bean>

    <bean id="PROPAGATION_REQUIRES_NEW" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW" />
    </bean>

    <bean id="PROPAGATION_SUPPORTS" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
        <property name="transactionManager" ref="txManager" />
        <property name="propagationBehaviorName" value="PROPAGATION_SUPPORTS" />
    </bean>

</blueprint>

A simple way of demonstrating that transaction policies have some effect on a transaction is to insert a PROPAGATION_NEVER policy into the middle of an existing transaction, as shown in the following route:

from("file:src/data?noop=true")
        .transacted()
        .bean("accountService","credit")
        .transacted("PROPAGATION_NEVER")
        .bean("accountService","debit");

Used in this way, the PROPAGATION_NEVER policy inevitably aborts every transaction, leading to a transaction rollback. You should easily be able to see the effect of this on your application.

The preceding route can be defined in Blueprint XML, as follows:

<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">

    <camelContext xmlns="http://camel.apache.org/schema/blueprint">
        <route>
            <from uri="file:src/data?noop=true" />
            <transacted />
            <bean ref="accountService" method="credit" />
            <transacted ref="PROPAGATION_NEVER" />
            <bean ref="accountService" method="debit" />
        </route>
    </camelContext>

</blueprint>

You can use one of the following approaches to roll back a transaction:

9.5.1.1. Using runtime exceptions to trigger rollbacks

The most common way to roll back a Spring transaction is to throw a runtime (unchecked) exception. In other words, the exception is an instance or subclass of java.lang.RuntimeException. Java errors, of java.lang.Error type, also trigger transaction rollback. Checked exceptions, on the other hand, do not trigger rollback.

The following figure summarizes how Java errors and exceptions affect transactions, where the classes that trigger roll back are shaded gray.

Note

The Spring framework also provides a system of XML annotations that enable you to specify which exceptions should or should not trigger roll backs. For details, see "Rolling back" in the Spring Reference Guide.

Warning

If a runtime exception is handled within the transaction, that is, before the exception has the chance to percolate up to the code that does the transaction demarcation, the transaction will not be rolled back. For details, see Section 9.5.2, “How to define a dead letter queue”.

from("file:src/data?noop=true")
    .transacted()
    .bean("accountService","credit")
    .choice().when(xpath("/transaction/transfer[amount > 100]"))
        .rollback()
    .otherwise()
        .to("direct:txsmall");

from("direct:txsmall")
    .bean("accountService","debit")
    .bean("accountService","dumpTable")
    .to("file:target/messages/small");

from("file:src/data?noop=true")
    .transacted()
    .bean("accountService","credit")
    .choice().when(xpath("/transaction/transfer[amount > 100]"))
        .markRollbackOnly()
    .otherwise()
        .to("direct:txsmall");
...

To keep track of failed transactions, you can define an onException() clause, which enables you to divert the relevant exchange object to a dead-letter queue. When used in the context of transactions, however, you need to be careful about how you define the onException() clause, because of potential interactions between exception handling and transaction handling. The following example shows the correct way to define an onException() clause, assuming that you need to suppress the rethrown exception.

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        onException(IllegalArgumentException.class)
            .maximumRedeliveries(1)
            .handled(true)
            .to("file:target/messages?fileName=deadLetters.xml&fileExist=Append")
            .markRollbackOnly();  // NB: Must come *after* the dead letter endpoint.

        from("file:src/data?noop=true")
            .transacted()
            .bean("accountService","credit")
            .bean("accountService","debit")
            .bean("accountService","dumpTable")
            .to("file:target/messages");
    }
}

In the preceding example, onException() is configured to catch the IllegalArgumentException exception and send the offending exchange to a dead letter file, deadLetters.xml. Of course, you can change this definition to catch whatever kind of exception arises in your application. The exception rethrow behavior and the transaction rollback behavior are controlled by the following special settings in the onException() clause:

Instead of using onException(), a simple approach to handling exceptions in a transactional route is to use the doTry() and doCatch() clauses around the route. For example, the following code shows how you can catch and handle the IllegalArgumentException in a transactional route, without the risk of getting trapped in an infinite loop.

// Java
import org.apache.camel.builder.RouteBuilder;

public class MyRouteBuilder extends RouteBuilder {
    ...
    public void configure() {
        from("file:src/data?noop=true")
            .doTry()
                .to("direct:split")
            .doCatch(IllegalArgumentException.class)
                .to("file:target/messages?fileName=deadLetters.xml&fileExist=Append")
            .end();

        from("direct:split")
            .transacted()
            .bean("accountService","credit")
            .bean("accountService","debit")
            .bean("accountService","dumpTable")
            .to("file:target/messages");
    }
}

In this example, the route is split into two segments. The first segment (from the file:src/data endpoint) receives the incoming exchanges and performs the exception handling using doTry() and doCatch(). The second segment (from the direct:split endpoint) does all of the transactional work. If an exception occurs within this transactional segment, it propagates first of all to the transacted() command, causing the current transaction to be rolled back, and it is then caught by the doCatch() clause in the first route segment. The doCatch() clause does not rethrow the exception, so the file endpoint does not do any retries and infinite looping is avoided.