The prototype of a transaction is an operation that conceptually consists of a single step (for example, transfer money from account A to account B), but must be implemented as a series of steps. Clearly, such operations are acutely vulnerable to system failures, because a crash is likely to leave some of the steps unfinished, leaving the system in an inconsistent state. For example, if you consider the operation of transferring money from account A to account B: if the system crashes after debiting account A, but before crediting account B, the net result is that some money disappears into thin air.

In order to make such an operation reliable, it must be implemented as a transaction. On close examination, it turns out that there are four key properties a transaction must have in order to guarantee reliable execution: these are the so-called ACID properties of a transaction.

The ACID properties of a transaction are defined as follows:

A transaction client is an API or object that enables you to initiate and end transactions. Typically, a transaction client exposes operations that enable you to begin, commit, or roll back a transaction. In the context of the Spring framework, the PlatformTransactionManager exposes a transaction client API.

Transaction demarcation refers to the initiating and ending of transactions (where transactions can be ended either by being committed or rolled back). Demarcation can be effected either explicitly (for example, by calling a transaction client API) or implicitly (for example, whenever a message is polled from a transactional endpoint).

A resource is any component of a computer system that can undergo a persistent or permanent change. In practice, a resource is almost always a database or a service layered over a database (for example, a message service with persistence). Other kinds of resource are conceivable, however. For example, an Automated Teller Machine (ATM) is a kind of resource: once a customer has physically accepted cash from the machine, the transaction cannot be reversed.

A transaction manager is responsible for coordinating transactions across one or more resources. In many cases, a transaction manager is built into a resource. For example, enterprise-level databases generally include a transaction manager that is capable of managing transactions involving that database. But for transactions involving more than one resource, it is normally necessary to employ an external transaction manager implementation.

For transactions involving a single resource, the transaction manager built into the resource can generally be used. For transactions involving multiple resources, however, it is necessary to use an external transaction manager or a transaction processing (TP) monitor. In this case, the resources must be integrated with the transaction manager by registering their XA switches. There is also an important difference between the types of algorithm that are used for committing single-resource systems and multiple-resource systems, as follows:

To understand transaction processing, it is crucial to appreciate the basic relationship between transactions and threads: transactions are thread-specific. That is, when a transaction is started, it is attached to a specific thread (technically, a transaction context object is created and associated with the current thread). From this point on (until the transaction ends), all of the activity in the thread occurs within this transaction scope. Conversely, activity in any other thread does not fall within this transaction's scope (although it might fall within the scope of some other transaction).

From this, we can draw a few simple conclusions:

A transaction context is an object that encapsulates the information needed to keep track of a transaction. The format of a transaction context depends entirely on the relevant transaction manager implementation. At a minimum, the transaction context contains a unique transaction identifier.

A distributed transaction refers to a transaction in a distributed system, where the transaction scope spans multiple network nodes. A basic prerequisite for supporting distributed transactions is a network protocol that supports transmission of transaction contexts in a canonical format (see also, Distributed transaction managers). Distributed transaction lie outside the scope of Fuse Mediation Router transactions.

The X/Open XA standard describes a standardized interface for integrating resources with a transaction manager. If you want to manage a transaction that includes more than one resource, it is essential that the participating resources support the XA standard. Resources that support the XA standard expose a special object, the XA switch, which enables transaction managers (or TP monitors) to take control of their transactions. The XA standard supports both the 1-phase commit protocol and the 2-phase commit protocol.

When it comes to choosing the products that implement your transaction system, there is a great variety of database products and transaction managers available, some free of charge and some commercial. All of them have nominal support for transaction processing, but there are considerable variations in the qualities of service supported by these products. This section provides a brief guide to the kind of features that you need to consider when comparing the reliability and sophistication of different transaction products.

The following features determine the quality of service of a resource:

ANSI SQL defines four transaction isolation levels, as follows:

Databases generally do not support all of the different transaction isolation levels. For example, some free databases support only READ_UNCOMMITTED. Also, some databases implement transaction isolation levels in ways that are subtly different from the ANSI standard. Isolation is a complicated issue, which involves trade offs with database performance (for example, see Isolation in Wikipedia).

In order for a resource to participate in a transaction involving multiple resources, it needs to support the X/Open XA standard. You also need to check whether the resource's implementation of the XA standard is subject to any special restrictions. For example, some implementations of the XA standard are restricted to a single database connection (which implies that only one thread at a time can process a transaction involving that resource).

The following features determine the quality of service of a transaction manager:

A key differentiator for transaction managers is the ability to support multiple resources. This normally entails support for the XA standard, where the transaction manager provides a way for resources to register their XA switches.

Some transaction managers support advanced capabilities for manipulating the associations between a transaction context and application threads, as follows:

Some transaction managers have the capability to manage transactions whose scope includes multiple nodes in a distributed system (where the transaction context is propagated from node to node using special protocols such as WS-AtomicTransactions or CORBA OTS).

Advanced transaction managers typically provide visual tools to monitor the status of pending transactions. This kind of tool is particularly useful after a system failure, where it can help to identify and resolve transactions that were left in an uncertain state (heuristic exceptions).

There are significant variations amongst transaction managers with respect to their robustness in the event of a system failure (crash). The key strategy that transaction managers use is to write data to a persistent log before performing each step of a transaction. In the event of a failure, the data in the log can be used to recover the transaction. Some transaction managers implement this strategy more carefully than others. For example, a high-end transaction manager would typically duplicate the persistent transaction log and allow each of the logs to be stored on separate host machines.

A transaction manager is the part of an application that is responsible for coordinating transactions across one or more resources. In the Spring framework, the transaction manager is effectively the root of the transaction system. Hence, if you want to enable transactions on a component in Spring, you typically create a transaction manager bean and pass it to the component.

The responsibilities of the transaction manager are as follows:

A local transaction manager is a transaction manager that can coordinate transactions over a single resource only. In this case, the implementation of the transaction manager is typically embedded in the resource itself and the Spring transaction manager is just a thin wrapper around this built-in transaction manager.

For example, the Oracle database has a built-in transaction manager that supports demarcation operations (using SQL operations, BEGIN, COMMIT, ROLLBACK, or using a native Oracle API) and various levels of transaction isolation. Control over the Oracle transaction manager can be exported through JDBC, which is how Spring is able to wrap this transaction manager.

It is important to understand what constitutes a resource, in this context. For example, if you are using a JMS product, the JMS resource is the single running instance of the JMS product, not the individual queues and topics. Moreover, sometimes, what appears to be multiple resources might actually be a single resource, if the same underlying resource is accessed in different ways. For example, your application might access a relational database both directly (through JDBC) and indirectly (through an object-relational mapping tool like Hibernate). In this case, the same underlying transaction manager is involved, so it should be possible to enrol both of these code fragments in the same transaction.

Of course, it is possible for an application to have many different local transaction managers working independently of each other. For example, you could have one route that manipulates JMS queues and topics, where the JMS endpoints reference a JMS transaction manager. Another route could access a relational database through JDBC. But you could not combine JDBC and JMS access in the same route and have them both participate in the same transaction.

A global transaction manager is a transaction manager that can coordinate transactions over multiple resources. In this case, you cannot rely on the transaction manager built into the resource itself. Instead, you require an external system, sometimes called a transaction processing monitor (TP monitor), that is capable of coordinating transactions across different resources.

The following are the prerequisites for global transactions:

Usually, a server connects directly to the resources involved in a transaction. In a distributed system, however, it is occasionally necessary to connect to resources that are exposed only indirectly, through a Web service or through a CORBA IDL interface. In this case, you require a TP monitor that is capable of supporting distributed transactions. Several standards are available that describe how to support transactions for various distributed protocols—for example, the WS-AtomicTransactions specification for Web services and the CORBA Object Transaction Service (OTS) specification for CORBA applications.

Figure 2.1 shows an overview of the Spring transaction architecture.

Figure 2.1. Spring Transaction Architecture

In the standalone deployment model, the Spring container provides access to persistent data sources and is responsible for managing the transactions associated with those data sources. A notable limitation of the standalone model is that the Spring container can support only local transaction managers, which means that only one data source (resource) at a time can participate in a transaction.

Spring supports a variety of different wrapper APIs for accessing persistent storage. For example, to access a database through JDBC, Spring provides the SimpleDriverDataSource class to represent the database instance and the JdbcTemplate class to provide access to the database using SQL. Wrappers are also provided for other kinds of persistent resource, such as JMS, Hibernate, and so on. The Spring data sources are designed to be compatible with the local transaction manager classes.

In Spring, a local transaction manager is a wrapper class that is responsible for managing the transactions of a single resource. The local transaction manager is responsible for starting, committing, and rolling back transactions. Typically, the way that you use a transaction manager in Fuse Mediation Router is that you pass the transaction manager reference to a transactional Fuse Mediation Router component bean.

Spring provides different types of local transaction manager for different types of data source. For example, Spring provides a DataSourceTransactionManager for JDBC, a JmsTransactionManager for JMS, a HibernateTransactionManager for Hibernate, and so on.

Figure 2.2 shows an overview of the OSGi transaction architecture in Fuse ESB. The core of the architecture is a JTA transaction manager based on Apache Geronimo, which exposes various transaction interfaces as OSGi services.

Figure 2.2. OSGi Transaction Architecture

The JTA Transaction Services Specification section of the OSGi Service Platform Enterprise Specification describes the kind of transaction support that can (optionally) be provided by an OSGi container. Essentially, OSGi mandates that the transaction service is accessed through the Java Transaction API (JTA).

The transaction service exports the following JTA interfaces as OSGi services (the JTA services):

Only one JTA provider should be made available in an OSGi container. In other words, the JTA services are registered only once and the objects obtained by importing references to the JTA services must be unique.

The Fuse ESB transaction service exports the following additional interfaces as OSGi services:

By obtaining a reference to the PlatformTransactionManager OSGi service, it is possible to integrate application bundles written using the Spring transaction API into the Fuse ESB transaction architecture.

For more details about the OSGi transaction architecture, see the introductory chapter from Deploying into the OSGi Container, in the Fuse ESB library.

The PlatformTransactionManager interface is the key abstraction in the Spring transaction API, providing the classic transaction client operations: begin, commit and rollback. This interface thus provides the essential methods for controlling transactions at run time.

Example 2.1 shows the definition of the org.springframework.transaction.PlatformTransactionManager interface.

package org.springframework.transaction;

public interface PlatformTransactionManager {
  TransactionStatus getTransaction(TransactionDefinition definition)
    throws TransactionException;

  void commit(TransactionStatus status) throws TransactionException;

  void rollback(TransactionStatus status) throws TransactionException;
}

The TransactionDefinition interface is used to specify the characteristics of a newly created transaction. It is used to specify the isolation level and the propagation policy of the new transaction. For more details, see Propagation Policies.

The TransactionStatus interface can be used to check the status of the current transaction (that is, the transaction associated with the current thread) and to mark the current transaction for rollback. It is defined as follows:

public interface TransactionStatus extends SavepointManager {
    boolean isNewTransaction();

    boolean hasSavepoint();

    void setRollbackOnly();

    boolean isRollbackOnly();

    void flush();

    boolean isCompleted();
}

The PlatformTransactionManager interface defines the following methods:

Generally, you do not use the PlatformTransactionManager interface directly. In Fuse Mediation Router, you typically use a transaction manager as follows:

This section provides a brief overview of all the transaction manager implementations provided by the Spring framework. In general, the implementations fall into two different categories: local transaction managers and global transaction managers.

Table 2.1 summarizes the local transaction manager implementations provided by the Spring framework. These transaction managers are distinguished by the fact that they support a single resource only.

Table 2.2 summarizes the global transaction manager implementations provided by the Spring framework. These transaction managers are distinguished by the fact that they can support multiple resources.

To enable transactions in a JMS component (thus enabling JMS endpoints to play the role either of a transactional resource or a transactional client), you need to:

In addition, you may want to adjust the JMS component's cache level setting. External transaction managers can impact caching performance.

The JmsComponent bean's id specifies the URI prefix used by JMS endpoints that will use the transactional JMS component. For example, in Example 3.1 the JmsComponent bean's id equals jmstx, so endpoint that use the configured JMS component use the jmstx: prefix.

The JmsConfiguration class supports a large number of other properties, which are essentially identical to the JMS URI options described in JMS in EIP Component Reference.

See JMS in EIP Component Reference for information on setting the cache level of the JMS component.

Example 3.1 shows the configuration of a JMS component, jmstx that supports Spring transactions. The JMS component is layered over an embedded instance of Apache ActiveMQ and the transaction manager is an instance of JmsTransactionManager.

<beans xmlns="http://www.springframework.org/schema/beans" 
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
  ...
  <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"/>
      <property name="cacheLevelName" value="CACHE_CONNECTION"/>
  </bean> 

  <bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
    <property name="connectionFactory" ref="jmsConnectionFactory" />
  </bean>
  
  <bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="vm://broker1?brokerConfig=xbean:tutorial/activemq.xml"/>
  </bean>

</beans>

To use this JMS component in a route you would use the URI prefix jmstx: as shown in Example 3.2.

from("jmstx:queue:rawStockQuotes")
    .process(myFormatter)
    .to("jmstx:queue:formattedStockQuotes");

The type of exchange created by a JMS consumer endpoint depends on the value of the JMSReplyTo header in the incoming message. If the JMSReplyTo header is absent from the incoming message, the consumer endpoint produces exchanges with the InOnly message exchange pattern (MEP). For example, consider the following route that receives a stream of stock quotes from the queue, queue:rawStockQuotes, reformats the incoming messages, and then forwards them to another queue, queue:formattedStockQuotes.

from("jmstx:queue:rawStockQuotes")
    .process(myFormatter)
    .to("jmstx:queue:formattedStockQuotes");

Routes that process InOnly exchanges can easily be combined with transactions. In the preceding example, the JMS queues are accessed through the transactional JMS instance, jmstx (see Configuring the JMS Component). The transaction initiated by the consumer endpoint, jmstx:queue:rawStockQuotes, ensures that each incoming message is reliably transmitted to the producer endpoint, jmstx:queue:formattedStockQuotes.

Typically, a route designed to work for InOnly exchanges does not work properly for InOut exchanges. Unfortunately, this leaves the route at the mercy of the external JMS client: if the client should accidentally set a JMSReplyTo header, the JMS consumer endpoint will create an InOut exchange, which could lead to errors in a route that is designed for InOnly exchanges.

To avoid the risk of creating InOut exchanges when they are not wanted, you can use the disableReplyTo option in the JMS consumer to enforce the InOnly MEP. For example, the following route is guaranteed to process all incoming messages as InOnly exchanges:

from("jmstx:queue:rawStockQuotes?disableReplyTo=true")
    .process(myFormatter)
    .to("jmstx:queue:formattedStockQuotes");

Figure 3.1 shows an overview of a scenario consisting of JMS consumer endpoint feeding into a route that ends with a JMS producer endpoint. This route is designed to process exclusively InOnly exchanges.

Figure 3.1. Transactional JMS Route that Processes InOnly Exchanges

Messages coming into the route shown in Figure 3.1 are processed as follows:

Combining InOut mode with transactional JMS endpoints is problematic. In most cases, this mode of operation is fundamentally inconsistent and it is recommended that you refactor your routes to avoid this combination.

In a JMS consumer endpoint, InOut mode is automatically triggered by the presence of a JMSReplyTo header in an incoming JMS message. In this case, the endpoint creates an InOut exchange to hold the incoming message and it will use the JMSReplyTo queue to send the reply message.

The InOut MEP is fundamentally incompatible with a route containing transactional JMS endpoints. In almost all cases, the route will hang and no reply will ever be sent. To understand why, consider the following route for processing payment requests:

from("jmstx:queue:rawPayments")
    .process(inputReformatter)
    .to("jmstx:queue:formattedPayments")
    .process(outputReformatter);

The JMS consumer endpoint, jmstx:queue:rawPayments, polls for messages, which are expected to have a JMSReplyTo header (for InOut mode). For each incoming message, a new transaction is started and an InOut exchange is created. After reformatting by the inputReformatter processor, the InOut exchange proceeds to the JMS producer endpoint, jmstx:queue:formattedPayments, which sends the message and expects to receive a reply on a temporary queue. This scenario is illustrated by Figure 3.2

Figure 3.2. Transactional JMS Route that Processes InOut Exchanges

If you want to implement a transactional JMS route that has request/reply semantics, the easiest solution is to refactor your route to avoid using InOut exchanges. The basic idea is that instead of defining a single route that combines a request leg and a reply leg, you should refactor it into two routes: one for the (outbound) request leg and another for the (inbound) reply leg. For example, the payments example could be refactored into two separate routes as follows:

from("jmstx:queue:rawPaymentsIn")
    .process(inputReformatter)
    .to("jmstx:queue:formattedPaymentsIn");

from("jmstx:queue:formattedPaymentsOut")
    .process(outputReformatter)
    .to("jmstx:queue:rawPaymentsOut");

Instead of a single incoming queue, queue:rawPayments, which uses the queue from JMSReplyTo for replies, we now have a pair of queues: queue:rawPaymentsIn, for receiving incoming requests, and queue:formattedPaymentsOut, for sending outgoing replies. Instead of a single outgoing queue, queue:formattedPayments, which implicitly uses a temporary queue for replies, we now have a pair of queues: queue:formattedPaymentsOut, for forwarding outgoing requests, and queue:formattedPaymentsIn, for receiving incoming replies. This scenario is illustrated by Figure 3.3.

Figure 3.3. Pair of Transactional JMS Routes that Support Request/Reply Semantics

There is a special case of a transactional JMS route where you can process InOut exchanges. If you look at the preceding examples, it is clear that the essential cause of deadlock in the route is the presence of JMS producer endpoints that obey request/reply semantics. In contrast to this, if you define a route where the JMS producer endpoints obey oneway semantics (fire-and-forget), deadlock does not occur.

For example, if you want to have a route that records all of the processed exchanges in a log queue, queue:log, you could define a route like the following:

from("jmstx:queue:inOutSource")
    .to(ExchangePattern.InOnly, "jmstx:queue:log")
    .process(myProcessor);

The exchanges coming into this route are of InOut type and both the consumer endpoint, jmstx:queue:inOutSource, and the producer endpoint, jmstx:queue:log, are transactional. The key to avoiding deadlock in this case is to force the producer endpoint to operate in oneway mode, by passing the ExchangePattern.InOnly parameter to the to() command,

To provide access to various kinds of persistent storage, Spring encapsulates the relevant API in a template class. The purpose of the template class is to provide a simplifying wrapper layer around each type of storage and to ensure that any required Spring features are integrated cleanly with the persistence layer.

Spring provides the following template classes for data access:

The org.springframework.jms.core.JmsTemplate class is a general-purpose class for managing Java Messaging Service (JMS) connections. One of the main advantages of this class is that it simplifies the JMS synchronous access codes.

To create an instance of a JmsTemplate, you need to supply a reference to a javax.jms.ConnectionFactory object.

The org.springframework.jdbc.core.JdbcTemplate class is a wrapper around a JDBC data source, enabling you to access a JDBC database using SQL operations.

To create an instance of a JdbcTemplate, you need to supply a reference to a javax.sql.DataSource object (for example, see JDBC Data Source).

The org.springframework.jdbc.core.simple.SimpleJdbcTemplate class is a convenience wrapper around the JdbcTemplate class. This class has been pared down so that it includes only the most commonly used template methods and it has been optimized to exploit Java 5 features.

The org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate class is a convenience wrapper around the JdbcTemplate class, which enables you to use named parameters instead of the usual ? placeholders embedded in a SQL statement.

The org.springframework.orm.ibatis.SqlMapClientTemplate class is a simplifying wrapper around the iBATIS SqlMapClient class. iBATIS is an Object Relational Mapper (ORM) that is capable of automatically instantiating Java objects based on a given SQL database schema.

The org.springframework.orm.hibernate3.HibernateTemplate class provides an alternative to working with the raw Hibernate 3 session API (based on sessions returned from SessionFactory.getCurrentSession()).

The org.springframework.orm.jdo.JdoTemplate class provides an alternative to working with the raw JDO PersistenceManager API. The main difference between the APIs relates to their exception handling. See the Spring JavaDoc for details.

The org.springframework.orm.jpa.JpaTemplate class provides an alternative to working with the raw JPA EntityManager API..

This section describes how to access a database through the Spring JdbcTemplate class and provides a code example that shows how to use the JdbcTemplate class in practice.

The org.springframework.jdbc.core.JdbcTemplate class is the key class for accessing databases through JDBC in Spring. It provides a complete API for executing SQL statements on the database at run time. The following kinds of SQL operations are supported by JdbcTemplate:

The JdbcTemplate query methods are used to send SELECT queries to the database. A variety of different query methods are supported, depending on how complicated the return values are.

The simplest case is where you expect the query to return a single value from a single row. In this case, you can use a type-specific query method to retrieve the single value. For example, if you want to retrieve the balance of a particular customer's account from the accounts table, you could use the following code:

// Java
int origAmount = jdbc.queryForInt(
    "select amount from accounts where name = ?",
    new Object[]{name}
);

The arguments to the SQL query are provided as a static array of objects, Object[]{name}. In this example, the name string is bound to the question mark, ?, in the SQL query string. If there are multiple arguments to the query string (where each argument in the SQL string is represented by a question mark, ?), you would provide an object array with multiple arguments—for example, Object[]{arg1,arg2,arg3,...}.

The next most complicated case is where you expect the query to return multiple values from a single row. In this case, you can use one of the queryForMap() methods to retrieve the contents of a single row. For example, to retrieve the complete account details from a single customer:

// Java
Map<String,Object> rowMap = jdbc.queryForMap(
    "select * from accounts where name = ?",
    new Object[]{name}
);

Where the returned map object, rowMap, contains one entry for each column, using the column name as the key.

The most general case is where you expect the query to return multiple values from multiple rows. In this case, you can use one of the queryForList() methods to return the contents of multiple rows. For example, to return all of the rows from the accounts table:

// Java
List<Map<String,Object> > rows = jdbc.queryForList(
    "select * from accounts"
);

In some cases, a more convenient way of returning the table rows is to provide a RowMapper, which automatically converts each row to a Java object. The return value of a query call would then be a list of Java objects. For example, the contents of the accounts table could be returned as follows:

// Java
List<Account> accountList = jdbc.query(
    "select * from accounts",
    new Object[]{},
    new RowMapper() {
        public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
            Account acc = new Account();
            acc.setName(rs.getString("name"));
            acc.setAmount(rs.getLong("amount"));
            return acc;
        }
    }
);

Where each Account object in the returned list encapsulates the contents of a single row.

The JdbcTemplate update methods are used to perform INSERT, UPDATE, or DELETE operations on the database. The update methods modify the database contents, but do not return any data from the database (apart from an integer return value, which counts the number of rows affected by the operation).

For example, the following update operation shows how to set the amount field in a customer's account:

// Java
jdbc.update(
    "update accounts set amount = ? where name = ?",
    new Object[] {newAmount, name}
);

For all other SQL operations, there is a general purpose execute() method. For example, you would use this method to execute a create table statement, as follows:

// Java
jdbc.execute("create table accounts (name varchar(50), amount int)");

To illustrate the database operations you can perform through the JdbcTemplate class, consider the account service, which provides access to bank account data stored in a database. It is assumed that the database is accessible through a JDBC data source and the account service is implemented by an AccountService class that exposes the following methods:

By combining credit and debit operations, it is possible to model money transfers, which can also be used to demonstrate key properties of transaction processing.

For the account service example, the money transfer orders have a simple XML format, as follows:

<transaction>
  <transfer>
    <sender>Major Clanger</sender>
    <receiver>Tiny Clanger</receiver>
    <amount>90</amount>
  </transfer>
</transaction>

When this money transfer order is executed, the amount of money specified in the amount element is debited from the sender account and credited to the receiver account.

Before we can start performing any queries on the database, the first thing we need to do is to create an accounts table and populate it with some initial values. Example 4.1 shows the definition of the CreateTable class, which is responsible for intializing the accounts table.

// Java
package com.fusesource.demo.tx.jdbc.java;

import javax.sql.DataSource;

import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;

public class CreateTable {
    private static Logger log = Logger.getLogger(CreateTable.class);

    protected DataSource dataSource;
    protected JdbcTemplate jdbc;

    public DataSource getDataSource() {
        return dataSource;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }
    
    public CreateTable(DataSource ds) {
        log.info("CreateTable constructor called");
        setDataSource(ds);
        setUpTable();
    }
    
    public void setUpTable() {
        log.info("About to set up table...");
        jdbc = new JdbcTemplate(dataSource);
        jdbc.execute("create table accounts (name varchar(50), amount int)");
        jdbc.update("insert into accounts (name,amount) values (?,?)",
                new Object[] {"Major Clanger", 2000}
        );
        jdbc.update("insert into accounts (name,amount) values (?,?)",
                new Object[] {"Tiny Clanger", 100}
        );
        log.info("Table created");
    }
}

<beans ...>
    <!-- datasource to the database -->
    <bean id="dataSource" class="org.springframework.jdbc.datasource.SimpleDriverDataSource">
        <property name="driverClass" value="org.hsqldb.jdbcDriver"/>
        <property name="url" value="jdbc:hsqldb:mem:camel"/>
        <property name="username" value="sa"/>
        <property name="password" value=""/>
    </bean>

    <!--  Bean to initialize table in the DB -->
    <bean id="createTable" class="com.fusesource.demo.tx.jdbc.java.CreateTable">
        <constructor-arg ref="dataSource" />
    </bean>
    ...
</beans>

Example 4.2 shows an outline of the AccountService class, not including the service methods that access the database. The class expects to receive a data source reference through dependency injection, which it then uses to create a JdbcTemplate instance.

package com.fusesource.demo.tx.jdbc.java;

import java.util.List;

import javax.sql.DataSource;


import org.apache.camel.Exchange;
import org.apache.camel.language.XPath;
import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;


public class AccountService {
    private static Logger log = Logger.getLogger(AccountService.class);
    private JdbcTemplate jdbc;

    public AccountService() {
    }

    public void setDataSource(DataSource ds) {
        jdbc = new JdbcTemplate(ds);
    }
    ...
    // Service methods (see below)
    ...
}

<beans ...>
    <!-- Bean for account service -->
    <bean id="accountService" class="com.fusesource.demo.tx.jdbc.java.AccountService">
        <property name="dataSource" ref="dataSource"/>
    </bean>
    ...
</beans>

The credit() method adds the specified amount of money, amount, to the specified account, name in the accounts database table, as follows:

    public void credit(
            @XPath("/transaction/transfer/receiver/text()") String name, 
            @XPath("/transaction/transfer/amount/text()") String amount
            )
    {
        log.info("credit() called with args name = " + name + " and amount = " + amount);
        int origAmount = jdbc.queryForInt( 
                "select amount from accounts where name = ?",
                new Object[]{name}
        );
        int newAmount = origAmount + Integer.parseInt(amount);
        
        jdbc.update( 
                "update accounts set amount = ? where name = ?",
                new Object[] {newAmount, name}
        );
    }

The debit() method subtracts the specified amount of money, amount, from the specified account, name in the accounts database table, as follows:

    public void debit(
            @XPath("/transaction/transfer/sender/text()") String name, 
            @XPath("/transaction/transfer/amount/text()") String amount
            )
    {
        log.info("debit() called with args name = " + name + " and amount = " + amount);
        int iamount = Integer.parseInt(amount);
        if (iamount > 100) { 
            throw new IllegalArgumentException("Debit limit is 100");
        }
        int origAmount = jdbc.queryForInt(
                "select amount from accounts where name = ?",
                new Object[]{name}
        );
        int newAmount = origAmount - Integer.parseInt(amount);
        if (newAmount < 0) { 
            throw new IllegalArgumentException("Not enough in account");
        }
        
        jdbc.update(
                "update accounts set amount = ? where name = ?",
                new Object[] {newAmount, name}
        );
    }

The dumpTable() method is convenient for testing. It simply returns the entire contents of the accounts table as a string. It is implemented as follows:

    public void dumpTable(Exchange ex) {
        log.info("dump() called");
        List<?> dump = jdbc.queryForList("select * from accounts");
        ex.getIn().setBody(dump.toString());
    }

Fuse Mediation Router provides a simple mechanism for initiating a transaction in a route, by inserting the transacted() command in the Java DSL or by inserting the <transacted/> tag in the XML DSL.

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

Figure 5.1. Demarcation by Marking the Route

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

// Java
import org.apache.camel.spring.SpringRouteBuilder;

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

In this example, the file endpoint reads some files in XML format that describe a transfer of funds from one account to another. The first beanRef() invocation credits the specified sum of money to the beneficiary's account and then the second beanRef() invocation subtracts the specified sum of money from the sender's account. Both of the beanRef() invocations cause updates to be made to a database resource, which we are assuming is bound to the transaction through the transaction manager (for example, see JDBC Data Source). For a sample implementation of the accountService bean, see Spring JDBC Template.

The beanRef() Java DSL command is available only in the SpringRouteBuilder class. It enables you to reference a bean by specifying the bean's Spring registry ID (for example, accountService). If you do not use the beanRef() command, you could inherit from the org.apache.camel.builder.RouteBuilder class instead.

The preceding route can equivalently be expressed in Spring XML, where the <transacted/> tag is used to mark the route as transactional, as follows:

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

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

</beans>

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 Spring 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 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 Sample route with PROPAGATION_NEVER policy in Java DSL.

If you insert a transacted processor into a route, a new transaction is created each time an exchange passes through this node and the transaction's scope is defined as follows:

In particular, all of the route nodes preceding the transacted processor are not included in the transaction (but the situation is different, if the route begins with a transactional endpoint—see Demarcation by Transactional Endpoints). For example, the following route is incorrect, because the transacted() DSL command mistakenly appears after the first beanRef() call (which accesses the database resource):

// Java
import org.apache.camel.spring.SpringRouteBuilder;

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

It is crucial to understand that a given transaction is associated with the current thread only. It follows that 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.spring.SpringRouteBuilder;

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

A route like 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.spring.SpringRouteBuilder;

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

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

        from("direct:txsmall")
            .beanRef("accountService","debit")
            .beanRef("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 Fuse Mediation Router 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. If you need to access multiple resources, you must deploy your application in a J2EE container, which gives you access to a global transaction manager.

For example, the following route sends the order for a money transfer to two different JMS queues: the credits queue processes the order to credit the receiver's account; and the debits queue processes the order to debit the sender's account. Since there must only be a credit, if there is a corresponding debit, it makes sense to enclose the enqueueing operations in a single transaction. If the transaction succeeds, both the credit order and the debit order will be enqueued, but if an error occurs, neither order will be enqueued.

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

If a consumer endpoint at the start of a route accesses a resource, the transacted() command is of no use, because it initiates the transaction after an exchange is polled. In other words, the transaction starts too late to include the consumer endpoint within the transaction scope. The correct approach in this case is to make the endpoint itself responsible for initiating the transaction. An endpoint that is capable of managing transactions is known as a transactional endpoint.

Figure 5.2 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.

Figure 5.2. Demarcation by Transactional Endpoints

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");

Where 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 will automatically attempt to redeliver the message).

The JMS component bean, jmstx, must be explicitly configured to use transactions, as follows:

<beans ...>
  <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>
  ...
</beans>

Where 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. For the complete Spring XML configuration of this component, see Example 3.1.

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

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

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

</beans>

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 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 Spring XML (see Default transaction manager and transacted policy)—so it is generally better not to include the transacted() DSL command in routes that start with a transactional endpoint.

The following Fuse Mediation Router 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.

If you want to influence the way a transactional client creates new transactions, you can do so by specifying a transaction policy for it. In particular, Spring transaction policies enable you to specify a propagation behavior for your transaction. For example, if a transactional client is about to create a new transaction and it detects that a transaction is already associated with the current thread, should it go ahead and create a new transaction, suspending the old one? Or should it simply let the existing transaction take over? These kinds of behavior are regulated by specifying the propagation behavior on a transaction policy.

Transaction policies are instantiated as beans in Spring XML. You can then reference a transaction policy by providing its bean ID as an argument to the transacted() DSL command. For example, if you want to initiate transactions subject to the behavior, PROPAGATION_REQUIRES_NEW, you could use the following route:

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

Where the PROPAGATION_REQUIRES_NEW argument specifies the bean ID of a transaction policy bean that is configured with the PROPAGATION_REQUIRES_NEW behavior (see Example 5.1).

Fuse Mediation Router 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:

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

The following propagation behaviors are supported by Spring (where these values were originally modelled on the propagation behaviors supported by J2EE):

Example 5.1 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.

<beans xmlns="http://www.springframework.org/schema/beans" 
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
       http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
  ...
  <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>

</beans>

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()
    .beanRef("accountService","credit")
    .transacted("PROPAGATION_NEVER")
    .beanRef("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 also be defined in Spring XML, as follows:

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" ... >

  <camelContext xmlns="http://camel.apache.org/schema/spring">
    <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>

</beans>

While you can use standard Fuse Mediation Router error handling techniques in a transactional route, it is important to understand the interaction between exceptions and transaction demarcation. In particular, you need to bear in mind that thrown exceptions usually cause transaction rollback.

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

The most common way to roll back a Spring transaction is to throw a runtime (unchecked) exception—that is, where 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. Figure 5.3 summarises how Java errors and exceptions affect transactions, where the classes that trigger rollback are shaded gray.

Figure 5.3. Errors and Exceptions that Trigger Rollback

If you want to trigger a rollback in the middle of a transacted route, you can do this by calling the rollback() DSL command, which throws an org.apache.camel.RollbackExchangeException exception. In other words, the rollback() command uses the standard approach of throwing a runtime exception to trigger the rollback.

For example, if you decide that there should be an absolute limit on the size of money transfers in the account services application, you could trigger a rollback when the amount exceeds 100, using the following code:

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

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

The markRollbackOnly() DSL command enables you to force the current transaction to roll back, without throwing an exception. This can be useful in cases where (as in Example 5.2) throwing an exception has unwanted side effects.

For example, Example 5.3 shows how to modify Example 5.2 by replacing rollback() with markRollbackOnly(). This version of the route solves the problem of the infinite loop. In this case, when the amount of the money transfer exceeds 100, the current transaction is rolled back, but no exception is thrown. Because the file endpoint does not receive an exception, it does not retry the exchange, and the failed transactions is quietly discarded.

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

In order 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. Example 5.4 shows the correct way to define an onException() clause, assuming that you need to suppress the rethrown exception.

// Java
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    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()
            .beanRef("accountService","credit")
            .beanRef("accountService","debit")
            .beanRef("accountService","dumpTable")
            .to("file:target/messages");
    }
}

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, Example 5.5 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.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {
    ...
    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()
            .beanRef("accountService","credit")
            .beanRef("accountService","debit")
            .beanRef("accountService","dumpTable")
            .to("file:target/messages");
    }
}