Chapter 6 covered ACID characteristics (atomicity, consistency, isolation, and durability) that Java and the Spring platform provide for transactional data sources (for example, relational databases) or JMS providers. Such a transaction against a single data source is often referred to as a local transaction. But a single data source is not where this support ends. Sometimes we can have requirements to ensure a single ACID transaction across various types of storage.

Java (and Spring abstractions) provide support for single transactions across multiple data sources. This is called a distributed transaction, or sometimes a global transaction. Such a transaction typically exceeds the boundaries of one server or network host and has to be synchronized across these data stores or JMS servers.

Understanding Distributed Transactions

So let’s say we have a PostgreSQL database and HornetQ JMS server. When we distribute the transaction on top of these stores, we can start reading messages from the HornetQ JMS destination and start inserting them via JDBC into the PostgreSQL storage. If one of the insert statements fails, we can be sure that our messages won’t get lost, because the transaction manager will ensure that the JMS operations performed in this global transaction are rolled back.

As you can imagine, this support is not trivial to implement; various mechanisms need to be in place for distributed transactions to be possible. First, a special global transaction manager needs to coordinate the distributed transaction across various transactional stores. This transaction manager needs to conform to the Java Transaction API (JTA). This API defines mechanisms for distributed transaction demarcation.

We also need to make sure that each store involved in the global transaction supports the XA standard (eXtended Architecture standard). This standard was developed by The Open Group (formerly X/Open), a consortium of several member organizations specializing in open Unix system standards (www.opengroup.org).

The XA standard specifies an interface between the global and local transaction managers to be able to participate in an ACID global transaction. An XA-compliant local transaction manager needs to have a JTA-compliant global transaction manager to cooperate with. You can think of XA as bridge between the local transaction manager (the manager handling transactions for a single data source) and the JTA global transaction manager (the manager coordinating local managers to cooperate in a global transaction). So, long story short, XA needs JTA in order to work properly. Cooperation between these standards is ensured via a two-phase commit algorithm/protocol, which works in the following phases:

1. Commit request phase (voting phase)
   1. The global transaction manager (also called the coordinator) asks all the XA local transaction managers (also called XA stores or participants) to decide whether a commit or rollback needs to be performed.
   2. If the participant decides to commit the transaction, the participant should then wait for orders from the global transaction manager and not perform any additional actions against the underlying storage.
   3. The participant may decide to roll back the transaction and disclose this intention to the global transaction manager.
2. Commit phase
   • d.   The global transaction manager decides, based on all the responses, whether all participants (local transaction managers) need to commit or roll back.
   • e.   The local transaction managers perform the chosen action (commit or rollback) that was ordered by the global transaction manager.

This algorithm involves various steps and can be hard to imagine. Figures 7-1 and 7-2 provide sequence diagrams for success and failure scenarios for this algorithm.

It is obvious that this algorithm involves a lot of ceremony to achieve the ACID transaction boundary across various data sources. We have to emphasize that the preceding figures depict scenarios with only two data sources. The amount of ceremony would multiply if we were to add more data sources.

It is important to remember that JTA is a standard that is part of Java Enterprise Edition. Therefore, various implementations of JTA and global transaction managers exist. These implementations can be stand-alone or part of a Java EE application server. In fact, if we are using Java EE transaction management embedded with an application server, we get this support out of the box.

So the list of JTA providers would include every Java EE–compliant application server. Additionally, the main stand-alone providers are as follows:

• Atomikos (www.atomikos.com)
• Bitronix JTA (https://github.com/bitronix/btm)

Cons of Distributed Transactions

Distributed transactions are powerful. We can imagine that it is able to magically solve all the possible problems (including duplicate data and lost data) described in the previous chapter. But this mechanism has significant performance overhead; if we bring into the equation a two-phase commit, which covers various data sources (two, three, or even more) and the various phases involved in it, the performance of this distributed transaction may be very slow.

This often means a huge problem for the high-performance requirements demanded in distributed systems. Therefore, the architects of these systems often try to avoid the need for two-phase commits. It is often much smarter to handle commits for separate data sources with the approach described at the end of Chapter 6, where we did an explicit check for duplicates. But sometimes requirements force us to use distributed transactions.

Reduced performance is the biggest, but not the only, problem of the two-phase commit algorithm. Too many parts are involved in such transactions, and the failure of any local transaction managers or global transaction managers can destroy the whole transaction. In addition, imagine a failure of a global transaction manager after the commit request phase, when all the local transaction managers can’t perform any actions and are waiting for him. They will keep waiting until a time-out appears (if there are any).

Another problem is monitoring the global transaction. If we want to troubleshoot, a common way is to log the sides involved in the communication. But three sides, at a minimum, are involved in the distributed transaction (a global and two local transaction managers), and each side creates at least one log entry for each phase. If logging is enabled for a production system, a lot of logging overhead is added, and processing slows even more.

So we can see there are a lot of reasons that we should strive to avoid distributed transactions.

Distributed Transactions with Spring

Configuring and implementing distributed transactions is not trivial. But we can be spared from this complexity by using Spring abstractions. This approach highly depends on understanding Spring transactional support.

This support is covered by a core module called spring-tx. It provides the generic interface PlatformTransactionManager, which enables transaction demarcation based on the @Transactional annotation. The spring-tx module is powerful and one of the main flagship features of Spring when we are dealing with transactional databases or JMS providers.

In fact, the same APIs are able to work with distributed transactions. Spring provides implementation of  JtaTransactionManager of the PlatformTransactionManager interface, which enables this support. But bear in mind that JtaTransactionManager itself doesn’t give us full JTA-compliant implementation of a global transaction manager. It is just a wrapper that operates on top of Java EE or a stand-alone JTA transaction manager implementation (for example, Atomikos or Bitronix JTA). So Spring alone doesn’t give us this support. This is smart, because why would Spring reinvent the wheel when it’s not necessary? Spring takes advantage of existing implementations and uses them under the JtaTransactionManager wrapper umbrella.

The main implication of this interface is that our transacted business code doesn’t need to know we are using a distributed transaction when it is marked as @Transactional. We just need to configure JtaTransactionManager, backed up by a JTA implementation and use various XA data sources to get ACID transaction characteristics across them.

Java EE servers have disadvantages when they force the presence of a JTA transaction manager on the platform. With Spring, we can simply change the configuration and turn this support on or off. The @Transactional annotation boundary specifies the start of a transaction, whether it is a global or local one. This will be seen in the examples of this chapter.

Common Classes for JTA Examples

To demonstrate ACID characteristics of distributed transactions across various data sources with Spring, we will apply a JTA transaction manager on examples similar to those in the preceding chapter that had problems with duplicating or losing messages. We will use a few common classes across all examples in this chapter. The first one, in Listing 7-1, is a repository sitting on the DAO layer.

package net.lkrnac.book.eiws.chapter07;

import javax.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;

@Repository
public class SimpleRepository {
  private final JdbcTemplate jdbcTemplate;

  @Autowired
  public SimpleRepository(JdbcTemplate jdbcTemplate) {
    super();
    this.jdbcTemplate = jdbcTemplate;
  }

  @PostConstruct
  public void initDbTable() {
    jdbcTemplate.execute("drop table TEXT_TABLE if exists");
    jdbcTemplate.execute("create table TEXT_TABLE(TEXT varchar(30))");
  }

  public void persistText(String text) {
    jdbcTemplate.update("insert into TEXT_TABLE values (?)", text);
  }
}

This class is similar to the DAO from Chapter 6. The only difference is a lack of querying for the number of stored text entries. The @Repository annotation marks this class as a Spring bean sitting on the persistence layer. For executing JDBC queries, it injects the JdbcTemplate instance via the constructor.

The initDbTable() method creates a fresh table after the Spring context is initialized. Its execution at this stage is ensured by the @PostConstruct annotation. This construct is obviously not typical for relational databases, but in this case we are using an in-memory database and want to have it empty for the examples, for each application to run.

The persistText() method is used to insert given text into the in-memory database. Listing 7-2 shows the service-layer class used for processing text messages.

package net.lkrnac.book.eiws.chapter07;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class SimpleService {
  private final SimpleRepository simpleRepository;

  @Autowired
  public SimpleService(SimpleRepository simpleRepository) {
    super();
    this.simpleRepository = simpleRepository;
  }

  public void processText(String text) {
    log.info("Process Message: {}", text);
    simpleRepository.persistText(text);
  }
}

Again this service is similar to that in the preceding chapter. The only difference is the missing method for reading the count of stored text records. The @Service annotation places this Spring bean at the service layer. The @Slf4j annotation is a Lombok feature for enabling easy access to the SLF4J logger. Constructor injection of the SimpleRepository instance enables use of the DAO bean instance.

Notice that we didn’t use the @Transactional annotation for any of these classes, because we plan to also include the JMS data source in the global transaction boundary. So the JMS data source will be used in components using SimpleService.

The processText() method is used in the examples to process text messages by logging the given message and persisting via the simpleRepository bean. The last class shared across the JTA examples is shown in Listing 7-3.

package net.lkrnac.book.eiws.chapter07;
import javax.jms.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SimpleMessageSender {
  private static final String SIMPLE_MESSAGE = "simple message";
  private final JmsTemplate jmsTemplate;

  @Autowired
  public SimpleMessageSender(
      @Qualifier("nonXaJmsConnectionFactory") ConnectionFactory nonXaConnectionFactory) {
    super();
    jmsTemplate = new JmsTemplate(nonXaConnectionFactory);
  }

  @Scheduled(initialDelay = 1000, fixedRate = Long.MAX_VALUE)
  public void send() {
    log.info("Sending message: {}", SIMPLE_MESSAGE);
    jmsTemplate.convertAndSend("ExpiryQueue", SIMPLE_MESSAGE);
  }
}

As you may notice, this sender is similar to that in the preceding chapter, with a few differences. First, we are not using the @PostConstruct annotation to trigger the sending of the message. This is because JTA configuration is sometimes not initiated fast enough after creating this SimpleMessageSender bean, so sending the message must wait for the configuration in this example. In a real-life application, this sending would be triggered by the external system consuming our application. Therefore, for this example, we can rely on scheduling with initialDelay, which will delay sending of the example message after context creation. The scheduling attribute fixedRate is configured to Long.MAX_VALUE to ensure that the second message won’t be sent during execution of the JTA example.

The second difference as compared to the example in the previous chapter, is injection of nonXaConnectionFactory instead of the jmsTemplate bean. This is needed because we want to have the sender of the message be independent of the JTA transaction. Therefore, in upcoming examples, we will create alongside the XA-compliant JMS connectionFactory, a nonXaConnectionFactory bean, which bypasses JTA ceremonies. Bear in mind that in real life this sender wouldn’t be placed in our message consumer application.

Other constructs should be familiar. The constructor injection of JmsTemplate enables use of the Spring bean for sending messages. The send() method itself first logs the fact that the message is being sent and performs this action.

Spring JTA Example with XML Configuration

Our first example of JTA integration with Spring is based mainly on XML configuration. There is no need to introduce a new XML namespace for it, because we operate in a non-JEE environment. If our application were deployed into a JEE container, we would use the jee namespace. First, we need to configure Atomikos as the JTA transaction manager for the Spring application. Listing 7-4 shows this XML configuration.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:tx="http://www.springframework.org/schema/tx"
  xsi:schemaLocation="http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/tx
    http://www.springframework.org/schema/tx/spring-tx.xsd">

  <tx:annotation-driven transaction-manager="transactionManager"/>

  <bean id="userTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager"
      init-method="init" destroy-method="close"/>

  <bean id="transactionManager"
      class="org.springframework.transaction.jta.JtaTransactionManager">
    <property name="transactionManager" ref="userTransactionManager"/>
    <property name="userTransaction" ref="userTransactionManager"/>
  </bean>
</beans>

In this configuration, we use the tx namespace to enable the @Transactional annotation under the umbrella of the JTA transaction manager. The JTA bean definition follows with use of the bean namespace. The first bean we need to create is com.atomikos.icatch.jta.UserTransactionManager as the global transaction manager. We already mentioned that Spring doesn’t have JTA capability built in, so we need help of an external JTA framework.

To integrate the Atomikos global transaction manager into the Spring application as PlatformTransactionManager, we need to wrap it into Spring’s JtaTransactionManager. This allows us to take advantage of Spring’s transaction JTA support on top of the Atomikos implementation. An important configuration step is creation of com.atomikos.icatch.jta.UserTransactionImp, which starts the Atomikos transaction service. Configuring this for the Java SE environment is important. For a Java EE drive application, we would use com.atomikos.icatch.jta.J2eeUserTransaction.

Listing 7-5 shows configuration of the JDBC data source.

<?xml version="1.0" encoding="UTF-8"?>
<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.xsd">

  <bean id="dataSource" class="com.atomikos.jdbc.AtomikosDataSourceBean">
    <property name="uniqueResourceName" value="jdbcDataSource" />
    <property name="xaDataSource">
      <bean class="org.h2.jdbcx.JdbcDataSource">
        <property name="url"
          value="jdbc:h2:mem:testdb"/>
        <property name="user" value="sa"/>
      </bean>
    </property>
  </bean>

  <bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
    <property name="dataSource" ref="dataSource"/>
  </bean>
</beans>

Here we use only the bean Spring XML namespace for defining Spring beans. First, we create an XA data source bean based on the com.atomikos.jdbc.AtomikosDataSourceBean implementation that Atomikos provides. This implementation is needed to properly work with the JTA global transaction manager. uniqueResourceName is a mandatory parameter, which needs to be unique in case we have various XA data sources of this type in the application. So the value jdbcDataSource is just fine for our example.

In the xaDataSource property, we wrap the local data source implementation. This wrapper is needed to allow the classical JDBC data source to participate in a global transaction. In this case, we are using an in-memory H2 database engine, as mentioned in the classical JDBC data source. We also need to wrap this bean into the JdbcTemplate bean to enable Spring JDBC support on top of this XA data source. So the JdbcTemplate doesn’t have a clue that it is dealing with an XA data source. Listing 7-6 shows the JMS configuration.

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:jee="http://www.springframework.org/schema/jee"
  xmlns:util="http://www.springframework.org/schema/util"
  xmlns:jms="http://www.springframework.org/schema/jms"
  xsi:schemaLocation="http://www.springframework.org/schema/jms
    http://www.springframework.org/schema/jms/spring-jms.xsd
    http://www.springframework.org/schema/jee
    http://www.springframework.org/schema/jee/spring-jee.xsd
    http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/util
    http://www.springframework.org/schema/util/spring-util.xsd">

  <util:map id="jndiProperties" map-class="java.util.Properties">
    <entry key="java.naming.factory.initial"
      value="org.jnp.interfaces.NamingContextFactory" />
    <entry key="java.naming.factory.url.pkgs"
      value="org.jboss.naming:org.jnp.interfaces" />
    <entry key="java.naming.provider.url" value="jnp://localhost:1099" />
  </util:map>

  <jee:jndi-lookup id=" " jndi-name="/ConnectionFactory"
    environment-ref="jndiProperties" />
  <jee:jndi-lookup id="connectionFactory" jndi-name="/XAConnectionFactory"
    environment-ref="jndiProperties" />
  <jee:jndi-lookup id="queue" jndi-name="/queue/ExpiryQueue"
    environment-ref="jndiProperties" />

  <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    <constructor-arg ref="connectionFactory" />
  </bean>

  <bean id="destinationResolver"
    class="org.springframework.jms.support.destination.JndiDestinationResolver">
    <property name="jndiEnvironment" ref="jndiProperties" />
  </bean>

  <jms:listener-container transaction-manager="transactionManager" acknowledge="transacted"
      destination-resolver="destinationResolver">
    <jms:listener destination="/queue/ExpiryQueue" ref="simpleMessageListener"
      method="readMessage" />
  </jms:listener-container>
</beans>

In this case, we use various Spring XML namespaces. The util namespace helps us define necessary JNDI properties for the HornetQ connection, and the jee namespace is needed for JNDI lookup of these properties. We already saw this configuration in the preceding chapter. But this example uses connector factory beans; notice that the default HornetQ configuration contains XAConfigurationFactory and also plain ConfigurationFactory. Use of XAConnectionFactory ensures that the connectionFactory JMS data source will be XA compliant and thus can participate in a global transaction. The nonXaJmsConnectionFactory bean is used by the SimpleMessageSender bean to send a message for the JTA example. This connection factory doesn’t participate in the JTA transaction, as it would confuse the example’s behavior.

The HornetQ JMS server has both connection factories specified in its default configuration. Chapter 6, explained how easy it is to download and start the HornetQ server. So if you are interested, the default configuration for XAConnectionFactory and ConnectionFactory can be found in the configuration file <hornet>/config/stand-alone/non-clustered/hornetq-jms.xml.

The next step is wrapping this connection factory into the JmsTemplate bean. This allows us to send (or synchronously receive) JMS messages. The DestinationResolver bean is needed to work with JNDI lookup properties to find the JMS destination we are planning to send and read from.

Last, we use the jms Spring namespace to configure the JMS listener plug-in’s JMS listener bean. Notice that we are using transactionManager as an attribute for this listener container. This allows the listener container to offer global transaction support. Another important attribute is enabling of transacted acknowledge mode in the acknowledge attribute. Without this configuration, the listener wouldn’t be transacted and therefore would ignore all transaction managers. This configuration is also the reason we don’t need to turn on <tx:transaction-driven> to enable the @Transactional annotation on SimpleMessageSender. We are not using this annotation in this case.

As the listener method is configured, readMessage() is implemented in the simpleMessageListener bean shown in Listing 7-7.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@Transactional
public class SimpleMessageListener {
  private SimpleService simpleService;
  private boolean errorSimulated = false;

  @Autowired
  public SimpleMessageListener(SimpleService simpleService) {
    super();
    this.simpleService = simpleService;
  }

  public void readMessage(String message) {
    simpleService.processText(message);
    postprocess(message);
  }

  private void postprocess(String message) {
    // simulate error
    if (!errorSimulated) {
      errorSimulated = true;
      throw new IllegalArgumentException(message);
    }
  }
}

The @Component annotation registers this class as a Spring bean. It injects the SimpleService bean via constructor injection. The readMessage() method is used for listening to JMS messages in the JMS configuration from Listing 7-6. The JMS listener container from this listing is configured to participate in a global transaction, so we have this bean covered by the @Transactional annotation, which is participating in the global JTA transaction (including JMS and the underlying JDBC data source).

This method first calls SimpleService to process the message, which means persisting it into the H2 in-memory database. After that, we simulate an error, but only for the first received message.

Notice that a similar implementation was already shown in Chapter 6, in Listing 6-28, as we discussed potential duplicates if an error occurs for a transacted JMS listener. In that case, the behavior was shown in Figure 6-9, and we ended up with a duplicate message being persisted in the database.

Listing 7-8 shows the main class of this example.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.boot.SpringApplication;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.transaction.annotation.EnableTransactionManagement;

@Configuration
@ComponentScan
@EnableJms
@EnableScheduling
@ImportResource(value = { "classpath:spring-jdbc-config.xml",
    "classpath:spring-jta-config.xml", "classpath:spring-jms-config.xml" })
public class JtaApplication {
  public static void main(String[] args) throws Exception {
    SpringApplication.run(JtaApplication.class, args);
  }
}

This class takes advantage of Spring Boot constructs to start the application based on the main method. It allows us to use this example as an executable JAR, something Spring Boot introduced. But we don’t use the autoconfiguration feature in this case, because we want to highlight plain Spring configuration with JTA.

Therefore, we have to enable each Spring feature separately:

• Component scanning in the current package and its subpackages by the @ComponentScan annotation
• Enabling Spring JMS via the @EnableJms annotation
• Enabling Spring scheduling support via the @EnableScheduling annotation, used for sending messages (see Listing 7-3)
• Importing all XML configurations explained for this example

After running this example, we can observe the output in Listing 7-9.

2015-09-05 16:44:08.260  INFO 4151 --- [           main] n.l.book.eiws.chapter07.JtaApplication   : Started JtaApplication in 1.905 seconds (JVM running for 2.582)
2015-09-05 16:44:08.272  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.BaseTransactionManager    : createCompositeTransaction ( 10000 ): created new ROOT transaction with id 127.0.1.1.tm0000100060
2015-09-05 16:44:09.254  INFO 4151 --- [pool-1-thread-1] n.l.b.e.chapter07.SimpleMessageSender    : Sending message: simple message
2015-09-05 16:44:09.343  INFO 4151 --- [erContainer#0-1] n.l.book.eiws.chapter07.SimpleService    : Process Message: simple message
2015-09-05 16:44:09.347  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AbstractDataSourceBean   : AtomikosDataSoureBean 'jdbcDataSource': getConnection ( null )
2015-09-05 16:44:09.348  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AbstractDataSourceBean   : AtomikosDataSoureBean 'jdbcDataSource': init...
2015-09-05 16:44:09.359  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : addParticipant ( XAResourceTransaction: 3132372E302E312E312E746D30303030313030303630:3132372E302E312E312E746D31 ) for transaction 127.0.1.1.tm0000100060
2015-09-05 16:44:09.359  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.start ( 3132372E302E312E312E746D30303030313030303630:3132372E302E312E312E746D31 , XAResource.TMNOFLAGS ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA
2015-09-05 16:44:09.360  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : registerSynchronization ( com.atomikos.jdbc.AtomikosConnectionProxy$JdbcRequeueSynchronization@e947e942 ) for transaction 127.0.1.1.tm0000100060
2015-09-05 16:44:09.361  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AtomikosConnectionProxy  : atomikos connection proxy for conn1: url=jdbc:h2:mem:testdb user=SA: calling prepareStatement(insert into TEXT_TABLE values (?))...
2015-09-05 16:44:09.370  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : setRollbackOnly() called for transaction 127.0.1.1.tm0000100060
2015-09-05 16:44:09.378  WARN 4151 --- [erContainer#0-1] o.s.j.l.DefaultMessageListenerContainer  : Execution of JMS message listener failed, and no ErrorHandler has been set.

org.springframework.jms.listener.adapter.ListenerExecutionFailedException: Listener method 'readMessage' threw exception; nested exception is java.lang.IllegalArgumentException: simple message
        at org.springframework.jms.listener.adapter.MessageListenerAdapter.invokeListenerMethod(MessageListenerAdapter.java:309)
        at org.springframework.jms.listener.adapter.MessageListenerAdapter.onMessage(MessageListenerAdapter.java:230)
        ...
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalArgumentException: simple message
        at net.lkrnac.book.eiws.chapter07.SimpleMessageListener.postprocess(SimpleMessageListener.java:28)
        at net.lkrnac.book.eiws.chapter07.SimpleMessageListener.readMessage(SimpleMessageListener.java:21)
        ...
        at org.springframework.jms.listener.adapter.MessageListenerAdapter.invokeListenerMethod(MessageListenerAdapter.java:301)
        ... 10 common frames omitted

2015-09-05 16:44:09.383  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AtomikosConnectionProxy  : atomikos connection proxy for conn1: url=jdbc:h2:mem:testdb user=SA: close()...
2015-09-05 16:44:09.384  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.end ( 3132372E302E312E312E746D30303030313030303630:3132372E302E312E312E746D31 , XAResource.TMSUCCESS ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA
2015-09-05 16:44:09.390  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.rollback ( 3132372E302E312E312E746D30303030313030303630:3132372E302E312E312E746D31 ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA
2015-09-05 16:44:09.398  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : rollback() done of transaction 127.0.1.1.tm0000100060
2015-09-05 16:44:09.411  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.BaseTransactionManager    : createCompositeTransaction ( 10000 ): created new ROOT transaction with id 127.0.1.1.tm0000200060
2015-09-05 16:44:09.429  INFO 4151 --- [erContainer#0-1] n.l.book.eiws.chapter07.SimpleService    : Process Message: simple message
2015-09-05 16:44:09.430  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AbstractDataSourceBean   : AtomikosDataSoureBean 'jdbcDataSource': getConnection ( null )...
2015-09-05 16:44:09.430  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AbstractDataSourceBean   : AtomikosDataSoureBean 'jdbcDataSource': init...
2015-09-05 16:44:09.430  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : addParticipant ( XAResourceTransaction: 3132372E302E312E312E746D30303030323030303630:3132372E302E312E312E746D32 ) for transaction 127.0.1.1.tm0000200060
2015-09-05 16:44:09.431  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.start ( 3132372E302E312E312E746D30303030323030303630:3132372E302E312E312E746D32 , XAResource.TMNOFLAGS ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA
2015-09-05 16:44:09.431  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : registerSynchronization ( com.atomikos.jdbc.AtomikosConnectionProxy$JdbcRequeueSynchronization@eafcc1e1 ) for transaction 127.0.1.1.tm0000200060
2015-09-05 16:44:09.431  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AtomikosConnectionProxy  : atomikos connection proxy for conn1: url=jdbc:h2:mem:testdb user=SA: calling prepareStatement(insert into TEXT_TABLE values (?))...
2015-09-05 16:44:09.433  INFO 4151 --- [erContainer#0-1] c.atomikos.jdbc.AtomikosConnectionProxy  : atomikos connection proxy for conn1: url=jdbc:h2:mem:testdb user=SA: close()...
2015-09-05 16:44:09.434  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.end ( 3132372E302E312E312E746D30303030323030303630:3132372E302E312E312E746D32 , XAResource.TMSUCCESS ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA
2015-09-05 16:44:09.434  INFO 4151 --- [erContainer#0-1] c.a.icatch.imp.CompositeTransactionImp   : commit() done (by application) of transaction 127.0.1.1.tm0000200060
2015-09-05 16:44:09.494  INFO 4151 --- [erContainer#0-1] c.a.datasource.xa.XAResourceTransaction  : XAResource.commit ( 3132372E302E312E312E746D30303030323030303630:3132372E302E312E312E746D32 , true ) on resource jdbcDataSource represented by XAResource instance xads0: conn0: url=jdbc:h2:mem:testdb user=SA

We can see various log entries from JTA and XA transaction managers, which demonstrates how resource intensive JTA transactions are. The behavior is best explained in Figure 7-3.

Notice that this sequence diagram covers processing of a single message. After reception of the message, it is inserted into TEXT_TABLE. But after the exception being raised in the postprocess() method of the JMS listener for the first received message, the global transaction manager will send a rollback order to the H2 database and to the JMS queue ExpiryQueue. This is different from the same implementation in Listener6-28, where the H2 database committed the local transaction.

After the rollback is sent to the JMS queue, the message is sent for a second round of processing. Again, the global transaction is started and the message is inserted into TEXT_TABLE. But on second reception, postprocess() doesn’t throw an exception, and our global transaction ends successfully, sending a commit to both data sources. Happy global ACID days. We can sleep well when our performance characteristics for our application using JTA are acceptable.

But notice that we are using a point-to-point messaging paradigm here. Dealing with JMS topics, where we would have various consumers, would involve an even more careful (and resource-intensive) approach to handle JTA transactions across each of them.

Spring JTA Example with Java Configuration

Every responsible Spring developer has noticed movement from XML to Java configuration. Therefore, this section covers how Spring JTA support can be configured without XML. Listing 7-10 shows a JTA configuration.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.jta.JtaTransactionManager;
import com.atomikos.icatch.jta.UserTransactionImp;
import com.atomikos.icatch.jta.UserTransactionManager;

@Configuration
@EnableTransactionManagement
public class JtaConfiguration {
  @Bean(initMethod = "init", destroyMethod = "close")
  public UserTransactionManager atomikosTransactionManager() {
    return new UserTransactionManager();
  }

  @Bean
  public JtaTransactionManager transactionManager(
      UserTransactionManager atomikosTransactionManager) {
    JtaTransactionManager transactionManager = new JtaTransactionManager();
    transactionManager.setTransactionManager(atomikosTransactionManager);
    transactionManager.setUserTransaction(new UserTransactionImp());
    return transactionManager;
  }
}

This configuration class enables Spring transactional support via the @EnableTransactionManagement annotation. As with XML configuration, we need to register the com.atomikos.icatch.jta.UserTransactionManager bean as an implementation of global transactional manager. We use the initMethod and destroyMethod attributes of this @Bean annotation to tell Spring how to instantiate and destroy this bean.

The second bean wraps the Atomikos transaction manager into JtaTransactionManager, which is an implementation of PlatformTransactionManager, and enables Spring transactional support for this application. Again we need to configure the userTransaction property to an instance of com.atomikos.icatch.jta.UserTransactionImp so that Atomikos will work fine for this non–Java EE application. Listing 7-11 dives into JDBC configuration of the XA data source.

package net.lkrnac.book.eiws.chapter07;

import javax.sql.DataSource;
import org.h2.jdbcx.JdbcDataSource;
import org.springframework.boot.jta.atomikos.AtomikosXADataSourceWrapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;

@Configuration
public class JdbcConfiguration {
  @Bean
  public DataSource dataSource() throws Exception {
    JdbcDataSource h2DataSource = new JdbcDataSource();
    h2DataSource.setUrl("jdbc:h2:mem:testdb");
    h2DataSource.setUser("sa");
    AtomikosXADataSourceWrapper wrapper = new AtomikosXADataSourceWrapper();
    return wrapper.wrapDataSource(h2DataSource);
  }

  @Bean
  public JdbcTemplate jdbcTemplate(DataSource dataSource) {
    return new JdbcTemplate(dataSource);
  }
}

Similar to JTA XML configuration, we need to wrap our H2 data source into AtomikosXADataSourceWrapper. This allows the H2 database to participate in global transactions. To enable use of handy Spring JDBC abstractions, we define the JdbcTemplate based on this XA data source. Listing 7-12 shows the JMS configuration.

package net.lkrnac.book.eiws.chapter07;

import java.util.Hashtable;
import javax.jms.ConnectionFactory;
import javax.jms.Session;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;

@Configuration
@EnableJms
public class JmsConfiguration {
  @Bean
  public InitialContext initialContext() throws NamingException {
    Hashtable<Object, Object> env = new Hashtable<Object, Object>();
    env.put("java.naming.factory.initial",
        "org.jnp.interfaces.NamingContextFactory");
    env.put("java.naming.factory.url.pkgs",
        "org.jboss.naming:org.jnp.interfaces");
    env.put("java.naming.provider.url", "jnp://localhost:1099");
    return new InitialContext(env);
  }

  @Bean
  public ConnectionFactory connectionFactory(InitialContext initialContext)
      throws NamingException {
    return (ConnectionFactory) initialContext.lookup("/XAConnectionFactory");
  }

  @Bean
  public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
    return new JmsTemplate(connectionFactory);
  }

  @Bean
  public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
      ConnectionFactory connectionFactory) {
    DefaultJmsListenerContainerFactory factory =
        new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setSessionTransacted(true);
    factory.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
    return factory;
  }
}

This configuration class enables JMS via the @EnableJms annotation and defines the InitialContext bean necessary for HornetQ access. The ConnectionFactory bean retrieves the XAConnectionFactory implementation from this HornetQ context. JmsTemplate and DefaultJmsListenerContainerFactory are standard Spring JMS beans explained in the previous chapter. The important configuration is enabling transacted mode of the listener container; otherwise, this listener container wouldn’t be plugged into the JTA infrastructure. The main class of this example is highlighted in Listing 7-13.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.boot.SpringApplication;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;

@Configuration
@ComponentScan
@EnableScheduling
public class JtaApplication {
  public static void main(String[] args) throws Exception {
    SpringApplication.run(JtaApplication.class, args);
  }
}

The main class uses the Spring Boot main method mechanism to start the application via the main method. Again, we avoid using autoconfiguration, as we want to highlight plain Spring configuration for JTA. @EnableScheduling turns on scheduling used by the message sender, as shown in Listing 7-13. @ComponentScan allows us to use Spring configurations and common classes we already described as well as the listener in Listing 7-14.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
@Transactional
public class SimpleMessageListener {
  private SimpleService simpleService;
  private boolean errorSimulated = false;

  @Autowired
  public SimpleMessageListener(SimpleService simpleService) {
    super();
    this.simpleService = simpleService;
  }

  @JmsListener(destination = "ExpiryQueue")
  public void readMessage(String message) {
    simpleService.processText(message);
    postprocess(message);
  }

  private void postprocess(String message) {
    // simulate error
    if (!errorSimulated) {
      errorSimulated = true;
      throw new IllegalArgumentException(message);
    }
  }
}

This listener is similar to the one in Listing 7-7, in the XML JTA example. The difference is that we have to specify that this bean is going to participate in a global transaction. This is done via the @Transactional annotation. The second difference is the @JmsListener annotation, which marks the method listening to the JMS messages. It is needed because there isn’t any separate XML configuration that would configure it for Spring.

When we run this example, we get exactly same behavior as that shown in Figure 7-1, and similar output as for the same example 0701-jta-xml-config in Listing 7-9. (I believe you don’t want to see that Mordor again.) But in this case, the configuration and code is more readable and maintainable than in the XML configuration example.

Spring Boot JTA Example

An even more readable and maintainable application can be based on Spring Boot autoconfiguration support, because it registers all the necessary configuration beans for us. We specify the HornetQ configuration in Listing 7-15.

spring.hornetq.mode=embedded
spring.hornetq.embedded.enabled=true
spring.hornetq.embedded.queues=ExpiryQueue

This configuration file enables an embedded instance of the HornetQ JMS server with a JMS destination called ExpiryQueue. The main class is shown in Listing 7-16.

package net.lkrnac.book.eiws.chapter07;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class JtaApplication {
  public static void main(String[] args) throws Exception {
    SpringApplication.run(JtaApplication.class, args);
  }
}

This Spring Boot main class uses the @SpringBootApplication annotation to enable a Spring Boot scan for opinionated configuration of the Spring application. It does a component scan of the current package plus subpackages and scans classpath dependencies to enable features provided by these libraries. In this case, it creates the necessary Spring infrastructure for the following:

• Atomikos JTA transaction manager
• Embedded HornetQ JMS server and JmsTemplate with XA support
• H2 embedded database and JdbcTemplate with XA support

These are defined in a Maven configuration for this example, which can be found on GitHub. And that’s all the configuration needed. It also uses common classes shown in Listing 7-1, 7-2, and 7-3 plus exactly the same implementation as the listener in Listing 7-14. This example nicely highlights why every Spring developer should look into Spring Boot, as it significantly reduces configuration for JTA support.

Again, when we run this example, the behavior is the same as in Figure 7-1. The same is true for the output, as you can see in Listing 7-9 (example 0701-jta-xml-config).

Summary

This chapter explored how JTA/XA protocols can help us cover various transactional data sources by global distributed transaction and enable ACID characteristics on top of them. Even when such support is compelling and easy to use with Spring, every smart developer should strive to avoid distributed transactions wherever possible.

Examples showed how easy it is to configure global transaction managers with XML and Java configuration and even easier with new mechanisms of Spring Boot. We also compared this support to plain transactional support and how it can help us target possible problems of using separate data sources within one application.