JMS Message Driven Channel Adapter + Transactional

[chudak]

I'm confused as to what you are doing...

You receive a message...it gets routed and sent to EITHER channel1 or channel2. It can't be sent to both so how can the same message get sent on channel1 and then throw an exception on channel2? Doesn't make any sense.

[srikanthradix]

I receive a message.
It gets ROUTED to both channel1 AND then channel2 (Router has multiple destinations).

If there is no exception it gets ROUTED to both channel1(which sends to Queue1) AND channel2(which sends to Queue2) and then acknowledgement is sent back. Message is removed from the Queue.

If there is an exception before ROUTING to channel2(to Queue2) and after it gets ROUTED to channel1(to Queue1), since the Queue1 has already consumed it, it is not getting removed from Queue1, Queue1 has already sent the message to its destination.

My question is how can I avoid the Queue1 consumption before transaction ends.

Transaction is like "If any exception occurs, everything should get rolledback right? Why not in this scenario?" How can I achieve this?

[chudak]

The problem is that typically transactions are bound to a thread. By putting the message on two separate channels, which are usually handled by separate threads, you are breaking the transaction semantics.

You'll have to review the documentation (or check the old posts) and see if someone has covered this concept of maintaining a single transaction across multiple concurrent channel executions.

It's possible that the recipient channel router handles this correctly..and that's actually closer to what you are actually doing.

[nos]

I have the exact same problem. JMS to JMS in a transaction (MQ). I have my own router with redirects the message to a predefined channel based on the content of the message. This channel then redirects it to the correct queue. If something fails in the inbound listener it rolls back fine but if something fails after that (i.e. target queue doesn't exist, channel doesn't exist etc) then the message is lost. I imagined that a session would exist between the inbound jms message driven channel and the outbound jms channel and if something would go wrong in-between that all would be rolled back nicely. Doesn't seem to be the case (even though atomikos, or so the log says/attempts so). I suppose the question is when does the session start and end in Spring Integration and how can we determine/influence it.... Of course it could just be that I haven't configured something correctly....

Here is some of my (almost complete) configuration....maybe it can help...

<!-- Poller -->
	<integration:poller id="poller" default="true" >
		<integration:interval-trigger interval="1000" />
		<integration:transactional
			transaction-manager="JtaTransactionManager" 
			propagation="REQUIRED"
			isolation="REPEATABLE_READ" 
			timeout="300" 
			read-only="false" 			
			/>
	</integration:poller>

	<!-- 1. Polls the input queue for incoming JMS messages -->
	<jms:message-driven-channel-adapter	id="jmsin" 
	destination="inputQueue" 
	channel="wltpContentRouterChannel" 
	transaction-manager="JtaTransactionManager"
	acknowledge="transacted" 
	 /> 

	<!-- 2. Channel for holding the input of the JMS poller. Note: direct channels must be used with transactions. -->
	<channel id="wltpContentRouterChannel">		
		<queue capacity="1"/>
	</channel>

	<!-- 3. Based on the header content of the xml text, the msg is redirected to the correct outbound handler -->
	<router ref="wltpContentRouter" input-channel="wltpContentRouterChannel" method="channelRerouter" ignore-channel-name-resolution-failures="false" />

	<!-- 4. All the possible channels 	-->
	<jms:outbound-channel-adapter id="FIN_PORTAL.BATCH-CHANNEL" destination="FINANCIAL_PORTAL.BATCH-QUEUE"  />
	<jms:outbound-channel-adapter id="VORBATCH-CHANNEL" destination="VORBATCH-QUEUE"  />

	<!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->
	<!-- Transaction management settings -->



	<!-- Construct Atomikos UserTransactionManager, needed to configure Spring -->
	<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>

	<!-- Also use Atomikos UserTransactionImp, needed to configure Spring -->
	<bean id="AtomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
		<property name="transactionTimeout" value="10000" />
	</bean>

	<!-- Configure the Spring framework to use JTA transactions from Atomikos -->
	<bean id="JtaTransactionManager"
		class="org.springframework.transaction.jta.JtaTransactionManager">
		<property name="transactionManager" ref="AtomikosTransactionManager" />
		<property name="userTransaction" ref="AtomikosUserTransaction" />
		<!-- This is required for Spring Batch set set a custom isolation level	when it runs -->		
		<property name="allowCustomIsolationLevels" value="true"></property>
	</bean>

	<!-- Configure the JMS connector; call init to register for recovery! -->
	<bean id="connectionFactory" class="com.atomikos.jms.AtomikosConnectionFactoryBean"
		init-method="init" destroy-method="close">
		<property name="uniqueResourceName" value="My_MQSeries_XA_RMI" />
		<property name="xaConnectionFactory" ref="xaConnectionFactory" />
		<!-- Adds it to a transaction I think... -->
		<property name="localTransactionMode" value="false"></property>
	</bean>



	<!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->