[Camel-users] ActiveMQ stuck message using JTA Transactions

Hi,

It is quite unclear why/what you are really trying to do.

First, When you say PROPAGATION_REQUIRED, you pass the transaction context
along the route hoping that somewhere along the way following proper
processing, the exchange will come back in a state where a commit/rollback
can be done on this transaction.

But halfway through, (Note you are using direct: an in-memory component),
you stomp the original transaction to create a new transaction without
sending back an success/fail on the original leg. This is what is causing
problems you are seeing, I believe.

If you create 2 distinct routes, not using direct but rather (http,
mina/netty, jms) then you should be able to create 2 separate and distinct
routes where the traction boundary does not carry forward beyond the
transport producer. Then I believe it should work.

Cheers,

Ashwin...

-----
---------------------------------------------------------
Ashwin Karpe
Apache Camel Committer & Sr Principal Consultant
FUSESource (a Progress Software Corporation subsidiary)
http://fusesource.com

Blog: http://opensourceknowledge.blogspot.com
---------------------------------------------------------

Thanks for your response. As mentioned I cleaned up the routes to be as
basic as possible to replicate the issue we are facing. Let me describe
why we have this type of routing and what we are trying to accomplish...

1. Start transaction 1
2. Consume message from queue foo
3. Perform some business logic to modify the message (Not shown in
sample routes)
4. Send off to direct:bar to continue routing
5. Suspend Transaction 1
6. Start a new Transaction 2
7. Perform some business logic to modify message (Not shown in sample
routes)
8. Insert message into a database (Not shown in sample routes)
9. Put message into queue bar
10. Commit Transaction 2
11. Resume Transaction 1
12. Commit Transaction 1

The reason for the multiple transactions is that if an error occurs
inside Transaction 2 we want to rollback whatever occurred in
Transaction 2 and we still want Transaction 1 to resume and then
eventually commit so the message is then finally removed from foo queue.
My understanding is that transactions are based on a single thread and
when using direct routing that will process the next route in the same
thread allowing use to suspend and resume transactions.

We used the example "Using multiple routes with different propagation
behaviors" describe on the Camel Website at
http://camel.apache.org/transactional-client.html as a basis for
implementing this routing.

It appears this issue arises when integrating with ActiveMQ, Spring JTA
TransactionManager and also Atomikos. I did see a reference to JMS
sending blocks identified on the Atomikos documentation
(http://www.atomikos.com/Documentation/KnownProblems#JMS_Sending_Blocks)
which describes a similar problem to what we are seeing happening. It
appears to fix that blocking you must set the "sessionTransaction"
attribute which we tried and it still does not work. I am posting on
camel to see if there is any issues with how the behind the scenes
consuming and producing from a queue is generated and how it integrates
with JTA transactions.

Included again for reference since the XML does not display in the
actual mailing list mails.

<camel:camelContext trace="true">
<camel:route>
<camel:from uri="activemq:queue:foo" />
<camel:transacted ref="PROPAGATION_REQUIRED"/>
<camel:to uri="direct:bar"/>
</camel:route>
<camel:route>
<camel:from uri="direct:bar" />
<camel:transacted ref="PROPAGATION_REQUIRES_NEW"/>
<camel:inOnly uri="activemq:queue:bar"/>
</camel:route>
</camel:camelContext>

<!-- define the activemq Camel component so we can integrate with the AMQ broker below -->
<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="transactionManager" ref="jtaTransactionManager"/>
<property name="transacted" value="true"/>
<property name="connectionFactory" ref="atomikosJmsConnectionFactory"/>
</bean>


<!-- JTA Transaction Manager Setup -->
<!-- this is the Spring JtaTransactionManager which under the hood uses Atomikos -->
<bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
<property name="transactionManager" ref="atomikosTransactionManager"/>
<property name="userTransaction" ref="atomikosUserTransaction"/>
</bean>

<!-- Is the ConnectionFactory to connect to the JMS broker -->
<!-- notice how we must use the XA connection factory -->
<bean id="jmsXaConnectionFactory" class="org.apache.activemq.ActiveMQXAConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>

<!-- Atomikos Setup -->
<!-- setup Atomikos for XA transaction -->
<bean id="atomikosTransactionManager" class="com.atomikos.icatch.jta.UserTransactionManager" init-method="init" destroy-method="close">
<!-- when close is called, should we force transactions to terminate or not? -->
<property name="forceShutdown" value="false"/>
</bean>

<!-- this is some atomikos setup you must do -->
<bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
<property name="transactionTimeout" value="300"/>
</bean>

<!-- this is some atomikos setup you must do -->
<bean id="atomikosJmsConnectionFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean" init-method="init" destroy-method="close">
<property name="uniqueResourceName" value="amq1"/>
<property name="xaConnectionFactory" ref="jmsXaConnectionFactory"/>
</bean>


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

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




*Brian Osborne | xFact, Inc.*
120 Water Street, Suite 214
North Andover, MA 01845
p. 978.686.3180 | c. 508.314.4627 | f. 978.824.2308
brian@xfact.com | www.xFact.com

You might be having issues with ActivMQ flow control.

I blogged about similar problem a while back. http://www.dragishak.com/?p=254

Basically, there is a limit on the total number of messages buffered in AMQ. In a transacted route a message will not be taken off the queue unless the transaction is successful. But you can't complete the transaction because, since all the resources are taken, you can't place the message on the next queue.


On Mar 14, 2011, at 12:43 PM, Brian wrote:

I am having an issue with a message being stuck in an ActiveMQ queue when
using spring JTA Transaction Manager backed by Atomikos and a propagation
policy of "PROPAGATION_REQUIRES_NEW".

I am trying to consume from one queue and insert into the second queue using
a different transaction than the first. This issue does not appear on the
first message but does on each subsequent message sent. We suspect there is
something wrong with the transactions and it is potentially not committing
successfully or hanging. Below is my stripped down camel configurations to
reproduce this issue. Thanks.

Camel Version - 2.6.0

ActiveMQ Version - 5.4.2

Spring Version - 3.0.5

Atomikos Version - 3.7.0

Interesting blog post.

However I don't think that is my issue since it can be reproduced by just
sending two messages to queue foo. The first message will go into queue foo
fine, get consumed and be put into queue bar. However the second message
will be consumed from queue foo be put into queue bar but will never fully
get consumed and removed from queue foo.

This only occurs after the first message and each subsequent message that
has been consumed from queue foo. However if we restart the system the
message that was stuck we will consumed from queue foo and be put into queue
bar which will then contain two copies of the same
message and then the next message will continue to be stuck in queue foo. It
appears that something is loosing context of the transaction/thread after
the first message causing the transaction/message to be blocking.

If I do ever get this working in this manner I will have to keep an eye out
for what you mentioned in your blog post since it will be used for a system
processing hundred of thousands of messages a day with periods of high
traffic so I will have to take a look into the flow control issue you
mentioned.

Thanks.

Brian Osborne | xFact, Inc.


Krsmanovic, Dragisa wrote: