DefaultMessageListenerContainer not receiving message when in running container

[PeteTh]

I have a problem which is probably related to transaction committing or using two instances of a jms backed channel (int-jms:channel x 2) in one pipeline - so an asynchronous process that spawns a 2nd asynchtonous process. I will describe the high-level flow (thread instance shown in braces) :-

1) [mainThread] Container receives a message from an external system via a custom spring integration inbound adapter. Spring Integration message created and moved to Channel that is jms backed (int-jms:channel). Like Message Dispatcher pattern.

2) [MessageListenerContainer] – receives message asynchronously from int-jms:channel

3) [MessageListenerContainer] Gateway used to act as a exception catcher - errorChannel defined

<int:service-activator input-channel="submission-jms" ref="errorHandlingGateway" />

<int:gateway id="errorHandlingGateway"
default-request-channel="submission-gw"
error-channel="submissionErrorChannel">
</int:gateway>

4) [MessageListenerContainer] Transformer of message

5) [MessageListenerContainer] recipient-list-router required to send message to 2 channels; 1st to 'Main' process/persist, 2nd to 'Ack' to send an acknowledgement message

Router list processing continues:-

'Main' processing :-
6) [MessageListenerContainer] Service Activator calling method that is @Transactional to persist message in database.

'Ack' processing :-
7) [MessageListenerContainer] Transformer to create Ack Message (payload is a String)

8) [MessageListenerContainer] int-jms:channel for 'AckJMSChannel'. This is used as this processing must be transaction aware (as pipeline will send a message to an external system that can't be recalled - so if commit fails the Ack pipeline must never start) By using jms it only becomes commited when [MessageListenerContainer] finally commits


Here's where the problems start.

9) [MessageListenerContainer-2nd-thread] – message should be received asynchronously

but this never happens.

I can see that the send to the JMS queue is successful by the logging from ActiveMQ

Initially I thought the transaction wasn't committing, but the database updates in step 6) are committing - as I can see data on database afterwards.

However, I don't know if the @Transactional on the Service/ServiceActivator is starting the transaction, rather than inheriting one from the DefaultMessageListenerContainer - which is what I was expecting.

We have not yet configured a XA Transaction Manager, but will do soon, however without this I thought if anything the commit would occur to the JMS 'AckJMSChannel' - and not be part of an XA Transaction, yet it doesn't seem to be committing, or the DefaultMessageListenerContainer is just not consuming.

What's also confusing is I have Unit Tested from step 9) 'AckJMSChannel' and here I note that:
- JUnit works fine if it is NOT @Transactional, [junit-thread] sends message to 'AckJMSChannel' and a [MessageListenerContainer] thread DefaultMessageListenerContainer.onMessage() gets called. JUnit has to wait for async processing to finish before closing. So the pipeline works.
- But if the JUnit is marked @Transactional then the same thing happens, DefaultMessageListenerContainer.onMessage never gets called, as if the message isn't commiting.

At the moment I believe I have tried both :-
acknowledge="transacted"
and alternatively
transaction-manager="transactionManager"
on the int-jms:channels

The total number of threads seems correct in that there seems to be instances [MessageListenerContainer] to match the total number of 'concurrency' attrbibute I have on the two int-jms:channels

Any thoughts would be much appreciated.

Note: I currently testing with ActiveMQ as JMS provider, but will be deployed to Weblogic (jms beans are varied with a Spring Profile)

[Gary Russell]

Best guess (it's not so easy without proper configuration and debug logs to look at).

Your errorHandlingGateway has no service-interface, so it is expecting (and won't receive) a reply. The default interface is RequestReplyExchanger.

Define an interface where the method returns void and thread won't be held up in the gateway awaiting a reply.