JMS with JTA using Spring

[seru]

Hi

I'm struggling to get global transactions to work with datasources and JMS.

I'm using JOTM with Enhydra for the distributed datasource bits and that all works as expected.

However, I also want to publish a message onto a topic in an ActiveMQ message server on another machine within the same transaction. I'm using JmsTemplate102 in a setup pretty similar to the Ch 12 examples in the Pro Spring book.

When I add a jmsTemplate.send(...) call into the business method, the message is sent OK but doesn't enlist itself with the current transaction. (The log messages show me the datasource being enlisted and bound onto the thread - but there isn't a similar message from the JMS package).

I've tried using org.activemq.ActiveMQXAConnectionFactory as the connection factory but that gives me the following exception...

• Caused by: javax.jms.JMSException: Session's XAResource has not been enlisted in a distributed transaction.
  at org.activemq.ActiveMQXASession.doStartTransaction( ActiveMQXASession.java:110)
  at org.activemq.ActiveMQSession.send(ActiveMQSession. java:1355)
  at org.activemq.ActiveMQMessageProducer.send(ActiveMQ MessageProducer.java:426)
  at org.activemq.ActiveMQMessageProducer.send(ActiveMQ MessageProducer.java:337)
  at org.activemq.ActiveMQTopicPublisher.publish(Active MQTopicPublisher.java:129)
  at org.springframework.jms.core.JmsTemplate102.doSend (JmsTemplate102.java:221)
  at org.springframework.jms.core.JmsTemplate.doSend(Jm sTemplate.java:762)
  at org.springframework.jms.core.JmsTemplate$2.doInJms (JmsTemplate.java:739)
  at org.springframework.jms.core.JmsTemplate.execute(J msTemplate.java:706)
  ... 44 more

I think I must be missing something but where do I tell the JMS packages about JOTM as I do with the datasources?

Here's the excerpts from my config files...

...
    <bean id="jotm" class="org.springframework.transaction.jta.JotmFactoryBean"/>

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

    <bean id="dataSource1" class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
        <property name="driverName">
            <value>oracle.jdbc.driver.OracleDriver</value>
        </property>
        <property name="transactionManager">
            <ref local="jotm"/>
        </property>
        <property name="url">
            <value>jdbc&#58;oracle&#58;thin&#58;@foo&#58;1521&#58;foo</value>
        </property>
    </bean>
    <bean id="dataSource" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
        <property name="dataSource">
            <ref local="dataSource1"/>
        </property>
        <property name="maxSize">
            <value>2</value>
        </property>
        <property name="minSize">
            <value>2</value>
        </property>
        <property name="user">
            <value>***</value>
        </property>
        <property name="password">
            <value>***</value>
        </property>
    </bean>

    <bean id="template" class="org.springframework.jdbc.core.JdbcTemplate">
      <constructor-arg><ref bean="dataSource"/></constructor-arg>
      <property name="nativeJdbcExtractor"><bean class="org.springframework.jdbc.support.nativejdbc.XAPoolNativeJdbcExtractor"/></property>
    </bean>

    <bean id="activeMQJndiTemplate" class="org.springframework.jndi.JndiTemplate">
        <property name="environment">
            <props>
                <prop key="java.naming.factory.initial">org.activemq.jndi.ActiveMQInitialContextFactory</prop>
                <prop key="brokerURL">tcp&#58;//foo&#58;61616</prop>
                <prop key="topic.SupportSystem.TicketUpdateTopic">SupportSystem.TicketUpdateTopic</prop>
            </props>
        </property>
    </bean>


    <bean id="connectionFactory" class="org.activemq.ActiveMQXAConnectionFactory">
        <property name="brokerURL">
            <value>tcp&#58;//foo&#58;61616</value>
        </property>
    </bean>

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate102">
        <property name="connectionFactory">
            <ref local="connectionFactory"/>
        </property>
        <property name="pubSubDomain"><value>true</value></property>
    </bean>

    <bean id="ticketUpdateTopic" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiTemplate">
            <ref bean="activeMQJndiTemplate"/>
        </property>
        <property name="jndiName">
            <value>SupportSystem.TicketUpdateTopic</value>
        </property>
    </bean>

I've read about the ActiveMQ JCA container, do I need to use this in order to specify a transaction manager? The Pro Spring book doesn't mention this at all which leaves me doubtful.

Thanks

Paul

[jstrachan]

If you are using a J2EE 1.4 container like Geronimo or JBoss 4, you could use the ActiveMQ Resource Adapter which integrates ActiveMQ into the JCA container of the J2EE application server.

Then you can look up the ConnectionFactory in JNDI (which will be the org.activemq.ra.ActiveMQConnectionFactory) which uses JCA Managed Connections to create a new JMS Connection.

Then you'd use the Spring JmsTemplate as usual - but this time the connection & session will automatically be registered in JTA for you.

So the easiest way of using JMS and JTA together is via JCA. However you can do it yourself with just Spring JmsTemplate - though you'll have to explicitly enlist the XASession with the JTA transaction yourself etc. So the JCA option is simpler

[seru]

Thanks for your reply James - I finally found some time to investigate this further...

We're using J2EE 1.3 at the moment so I've had a look at doing the transaction management manaually as you suggested.

I can easily manually enlist the JMS code with the current JTA transaction in our business method - ie. the transaction created by Spring via the TransactionProxyFactoryBean. This looks something like:

// Do some Ibatis based Jdbc stuff

            XATopicConnection connection = topicConnectionFactory.createXATopicConnection();
            XATopicSession session = connection.createXATopicSession();
            XAResource jmsXA = session.getXAResource();
            transactionManager.getTransactionManager().getTransaction().enlistResource(jmsXA);

            Message message = session.createTextMessage("Foo");
            TopicPublisher publisher = ((TopicSession) session).createPublisher(ticketUpdateTopic);
            publisher.publish(message);

// Do some more stuff

This all works fine - the JMS joins in with the transaction and gets rolled back when exceptions occur elsewhere.

However, in your post you mentioned that I could also use JmsTemplate with the manual solution. I can't see any way of getting the XA resources I need via this class - it appears to just to be able to do local transactions via a JmsTransactionManager. Am I missing something?

Thanks

Paul

[adepue]

We've recently had to solve this issue. We are using Spring in Tomcat, and so do not have all the features of a J2EE container - yet, we need transactional support for our (ActiveMQ) JMS producers. There are currently some JCA container projects underway for Spring (so that Spring can look like a JCA container), but they are not yet mature enough for JMS producers (at least, not for our needs). As a temporary solution, I've written a PooledSpringXAConnectionFactory for ActiveMQ that automatically handles pooling connections, sessions, and producers while also automatically enlisting in any current JTA transaction. Thread safety is maintained in the pooling implementation, and the same underlying XASession will be used per-thread + per-transaction (so you don't have a bunch of pooled XA resources being enlisted against the transaction). You can grab the source along with some instructions from the ActiveMQ JIRA: http://jira.logicblaze.com/jira/browse/AMQ-303

- Andy

[adepue]

Let me also comment that because JMSTemplate opens a new connection+session (and then closes them) for each invocation, then it will work well with the PoolingSpringXAConnectionFactory.

- Andy