we have a problem with Atomikos (TransactionManager) and ActiveMQ. Atomikos is used in a Spring environment to enable XA transactions. We tested the failover behavior of the ActiveMQ client side and notice some exceptions that are not going away. Our test scenario is the following:
We use the following versions:
Now we get the following exception from Atomikos:
2014-06-12 11:28:42 tpe-1 com.atomikos.datasource.xa.XAResourceTransaction WARN - XA resource 'connectionFactoryOut': commit for XID '6170695F3134303235363533313236323230303334383030303031:6170695F31343032353635333132363232333438' raised -7: the XA resource has become unavailable
javax.transaction.xa.XAException: The JMS connection has failed: java.io.EOFException
at org.apache.activemq.TransactionContext.toXAException(TransactionContext.java:793)
at org.apache.activemq.TransactionContext.commit(TransactionContext.java:590)
at com.atomikos.datasource.xa.XAResourceTransaction.commit(XAResourceTransaction.java:733)
at com.atomikos.icatch.imp.CommitMessage.send(CommitMessage.java:72)
at com.atomikos.icatch.imp.PropagationMessage.submit(PropagationMessage.java:83)
at com.atomikos.icatch.imp.Propagator$PropagatorThread.run(Propagator.java:79)
at com.atomikos.icatch.imp.Propagator.submitPropagationMessage(Propagator.java:58)
at com.atomikos.icatch.imp.CoordinatorStateHandler.commitFromWithinCallback(CoordinatorStateHandler.java:582)
at com.atomikos.icatch.imp.ActiveStateHandler$6.doCommit(ActiveStateHandler.java:301)
at com.atomikos.icatch.imp.CoordinatorStateHandler.commitWithAfterCompletionNotification(CoordinatorStateHandler.java:852)
at com.atomikos.icatch.imp.ActiveStateHandler.commit(ActiveStateHandler.java:296)
at com.atomikos.icatch.imp.CoordinatorImp.commit(CoordinatorImp.java:707)
at com.atomikos.icatch.imp.CoordinatorImp.terminate(CoordinatorImp.java:968)
at com.atomikos.icatch.imp.CompositeTerminatorImp.commit(CompositeTerminatorImp.java:82)
at com.atomikos.icatch.imp.CompositeTransactionImp.commit(CompositeTransactionImp.java:336)
at com.atomikos.icatch.jta.TransactionImp.commit(TransactionImp.java:190)
at com.atomikos.icatch.jta.TransactionManagerImp.commit(TransactionManagerImp.java:436)
at com.atomikos.icatch.jta.UserTransactionImp.commit(UserTransactionImp.java:107)
at org.springframework.transaction.jta.JtaTransactionManager.doCommit(JtaTransactionManager.java:1021)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:757)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:726)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:150)
at net.sprd.messaging.test.data_generator.SenderThread$1.run(SenderThread.java:42)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.activemq.ConnectionFailedException: The JMS connection has failed: java.io.EOFException
at org.apache.activemq.ActiveMQConnection.checkClosedOrFailed(ActiveMQConnection.java:1483)
at org.apache.activemq.TransactionContext.commit(TransactionContext.java:551)
... 26 more
Caused by: java.io.EOFException
at java.io.DataInputStream.readInt(DataInputStream.java:392)
at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:275)
at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:221)
at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:213)
at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
... 1 more
2014-06-12 11:28:42 tpe-1 com.atomikos.icatch.imp.CommitMessage WARN - Unexpected error in commit
com.atomikos.icatch.SysException: XA resource 'connectionFactoryOut': commit for XID '6170695F3134303235363533313236323230303334383030303031:6170695F31343032353635333132363232333438' raised -7: the XA resource has become unavailable
at com.atomikos.datasource.xa.XAResourceTransaction.commit(XAResourceTransaction.java:773)
at com.atomikos.icatch.imp.CommitMessage.send(CommitMessage.java:72)
at com.atomikos.icatch.imp.PropagationMessage.submit(PropagationMessage.java:83)
at com.atomikos.icatch.imp.Propagator$PropagatorThread.run(Propagator.java:79)
at com.atomikos.icatch.imp.Propagator.submitPropagationMessage(Propagator.java:58)
at com.atomikos.icatch.imp.CoordinatorStateHandler.commitFromWithinCallback(CoordinatorStateHandler.java:582)
at com.atomikos.icatch.imp.ActiveStateHandler$6.doCommit(ActiveStateHandler.java:301)
at com.atomikos.icatch.imp.CoordinatorStateHandler.commitWithAfterCompletionNotification(CoordinatorStateHandler.java:852)
at com.atomikos.icatch.imp.ActiveStateHandler.commit(ActiveStateHandler.java:296)
at com.atomikos.icatch.imp.CoordinatorImp.commit(CoordinatorImp.java:707)
at com.atomikos.icatch.imp.CoordinatorImp.terminate(CoordinatorImp.java:968)
at com.atomikos.icatch.imp.CompositeTerminatorImp.commit(CompositeTerminatorImp.java:82)
at com.atomikos.icatch.imp.CompositeTransactionImp.commit(CompositeTransactionImp.java:336)
at com.atomikos.icatch.jta.TransactionImp.commit(TransactionImp.java:190)
at com.atomikos.icatch.jta.TransactionManagerImp.commit(TransactionManagerImp.java:436)
at com.atomikos.icatch.jta.UserTransactionImp.commit(UserTransactionImp.java:107)
at org.springframework.transaction.jta.JtaTransactionManager.doCommit(JtaTransactionManager.java:1021)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:757)
at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:726)
at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:150)
at net.sprd.messaging.test.data_generator.SenderThread$1.run(SenderThread.java:42)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
The Spring configuration for the sending parts looks like this:
<bean id="jmsTemplateBroker" class="org.springframework.jms.core.JmsTemplate">
<constructor-arg name="connectionFactory" ref="cachingSendingSessionFactoryBroker" />
<property name="sessionTransacted" value="true" />
</bean>
<bean id="cachingSendingSessionFactoryBroker" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg name="targetConnectionFactory" ref="atomikosConnectionFactoryBroker" />
<property name="sessionCacheSize" value="1000" />
<property name="cacheConsumers" value="false" />
<property name="cacheProducers" value="false" />
</bean>
<bean id="atomikosConnectionFactoryBroker" class="com.atomikos.jms.AtomikosConnectionFactoryBean"
init-method="init" destroy-method="close" depends-on="activeMQConnectionFactoryBroker">
<property name="uniqueResourceName" value="atomikosConnectionFactoryBroker" />
<property name="xaConnectionFactory" ref="activeMQConnectionFactoryBroker" />
<property name="minPoolSize" value="1" />
<property name="maxPoolSize" value="1000" />
</bean>
<bean id="activeMQConnectionFactoryBroker" class="org.apache.activemq.ActiveMQXAConnectionFactory">
<property name="brokerURL" value="${broker.brokerURL}" />
<property name="userName" value="testUser" />
<property name="password" value="testPassword" />
<property name="redeliveryPolicy" ref="redeliveryPolicyBroker" />
<property name="prefetchPolicy" ref="activeMQPrefechPolicyBroker" />
</bean>
<bean id="activeMQPrefechPolicyBroker" class="org.apache.activemq.ActiveMQPrefetchPolicy">
<property name="all" value="100" />
</bean>
<bean id="redeliveryPolicyBroker" class="org.apache.activemq.RedeliveryPolicy">
<property name="initialRedeliveryDelay" value="1000" />
<property name="redeliveryDelay" value="1000" />
<property name="useExponentialBackOff" value="false" />
<property name="maximumRedeliveries" value="1000" />
</bean>
The Atomikos beans are configured as follows:
<bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager"
init-method="init" destroy-method="close" depends-on="userTransactionService">
<property name="forceShutdown" value="true" />
</bean>
<bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
<property name="transactionTimeout" value="70" />
</bean>
<bean id="timestamp" class="java.lang.String">
<constructor-arg value="#{'' + T(java.lang.System).currentTimeMillis()}" />
</bean>
<bean id="userTransactionService" class="com.atomikos.icatch.config.UserTransactionServiceImp"
init-method="init" destroy-method="shutdownForce">
<constructor-arg>
<props>
<prop key="com.atomikos.icatch.service">com.atomikos.icatch.standalone.UserTransactionServiceFactory</prop>
<prop key="com.atomikos.icatch.output_dir">/tmp/atomikosOutput</prop>
<prop key="com.atomikos.icatch.console_file_name">/tmp/api_#{timestamp}.out</prop>
<prop key="com.atomikos.icatch.console_log_level">DEBUG</prop>
<prop key="com.atomikos.icatch.tm_unique_name">api_#{timestamp}</prop>
<prop key="com.atomikos.icatch.threaded_2pc">false</prop>
<prop key="com.atomikos.icatch.max_actives">-1</prop>
<prop key="com.atomikos.icatch.max_timeout">10000</prop>
<prop key="com.atomikos.icatch.default_jta_timeout">999</prop>
<prop key="com.atomikos.icatch.serial_jta_transactions">false</prop>
<prop key="com.atomikos.icatch.enable_logging">false</prop>
<prop key="com.atomikos.icatch.log_base_name">api#{timestamp}</prop>
<prop key="com.atomikos.icatch.log_base_dir">/tmp/atomikos</prop>
<prop key="com.atomikos.icatch.checkpoint_interval">5000</prop>
</props>
</constructor-arg>
</bean>
These exceptions are thrown periodically. We assume the following: Atomikos sends a prepare statement to the ActiveMQ broker. After that the broker is shut down. Now the prepared statement is lost at the ActiveMQ side and so Atomikos will try to recover this gone transaction. Because both ActiveMQ broker use the same database we think that the prepared statement should be stored in the database. So the slave broker can resume the transaction. But this doesn't happen.
Has anyone an idea if this is a bug of ActiveMQ or if there is a "hidden" configuration property of ActiveMQ to enable the prepared mode of XA transactions?
Side mark: we make sure that we use unique ids for the transaction manager name (related to: Stackoverflow article about Atomikos unique ids)
Thanks.
This apparently is/was a bug. It's fixed with version 3.9.9 (even though this version/fix is currently only available for Atomikos customers).