JMS Reply To Issue

[Richard Holt]

Hi,

I have spent the last few days investigating an issue with JMS and ReplyTo. Just a little background, we are using 2 JMS brokers, one local and one remote and we were seeing issues whereby an error was raised where a Mule consuming from the remote queue was trying to respond to a Temporary Reply To set on the producing Mule system. This is running Mule 2.1.1 and Active MQ 5.1.0.

My conclusion is that the following is happening :

Mule Server A published to a queue using remoteSync = true. This sets the JMSReplyTo header on the message.
Mule Server B sees the JMSReplyTo and treats this as requiring a response and instantiates the JMSReplyToHandler. In the meantime however on Mule Server A the consumer.receive in the JMSMessageDispatcher has consumed from the temporary JMS queue and deleted the temporary queue. Mule Server B will then continue and write to the temporary queue which has been deleted.

Here is a test configuration file that shows the problem. Also i have included the code for the JmsRecipientList below.

To get this test to work you have to put the JmsReplyToHandler.java into your code and either put a breakpoint into it or a timer at the start of "processReplyTo" which halts the system for longer than 3 seconds!!

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesource.org/schema/mule/core/2.1"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:spring="http://www.springframework.org/schema/beans"
        xmlns:jms="http://www.mulesource.org/schema/mule/jms/2.1"
        xmlns:jbossts="http://www.mulesource.org/schema/mule/jbossts/2.1"
        xmlns:file="http://www.mulesource.org/schema/mule/file/2.1"
        xmlns:management="http://www.mulesource.org/schema/mule/management/2.1"
        xmlns:context="http://www.springframework.org/schema/context"
        xsi:schemaLocation="http://www.mulesource.org/schema/mule/core/2.1 http://www.mulesource.org/schema/mule/core/2.1/mule.xsd
                http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.1.xsd
                http://www.mulesource.org/schema/mule/jms/2.1 http://www.mulesource.org/schema/mule/jms/2.1/mule-jms.xsd
                http://www.mulesource.org/schema/mule/management/2.1 http://www.mulesource.org/schema/mule/management/2.1/mule-management.xsd
                http://www.mulesource.org/schema/mule/jbossts/2.1 http://www.mulesource.org/schema/mule/jbossts/2.1/mule-jbossts.xsd   
                http://www.mulesource.org/schema/mule/file/2.1 http://www.mulesource.org/schema/mule/file/2.1/mule-file.xsd">
               
        <file:connector name="fileConnector" autoDelete="true" streaming="false"/>

        <jms:activemq-connector name="localJmsConnector"
                acknowledgementMode="AUTO_ACKNOWLEDGE" recoverJmsConnections="true" persistentDelivery="true">
        </jms:activemq-connector>

        <model name="testModel">
                <service name="testService">
                  <inbound>
                    <file:inbound-endpoint path="c:/data/in" moveToDirectory="c:/save"/>
                  </inbound>
                  <outbound>
                        <custom-outbound-router class="com.acme.JmsRecipientList"/>
                  </outbound>
                </service>

                <service name="testService2">
                  <inbound>
                    <jms:inbound-endpoint queue="test2.queue" connector-ref="localJmsConnector"/>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                                <file:outbound-endpoint path="c:/data/out2" connector-ref="fileConnector"/>
                  </pass-through-router>
                  </outbound>
                </service>

                <service name="testService3">
                  <inbound>
                    <jms:inbound-endpoint queue="test3.queue" connector-ref="localJmsConnector"/>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                                <file:outbound-endpoint path="c:/data/out3" connector-ref="fileConnector"/>
                  </pass-through-router>
                  </outbound>
                </service>
        </model>
</mule>

JmsRecipientList.java

package com.acme;

import java.util.ArrayList;
import java.util.List;

import org.mule.api.MuleMessage;
import org.mule.routing.outbound.AbstractRecipientList;

public class JmsRecipientList extends AbstractRecipientList {

        @Override
        protected List<?> getRecipients(MuleMessage muleMessage) {
                List<String> result = new ArrayList<String>();
               
                result.add("jms://test2.queue?remoteSync=true");
                result.add("jms://test3.queue");
               
                return result;
        }
}




Thanks in advance for any assistance, this is a business critical issue for us.

Regards,

Richard

[Andrew Perepelytsya]

In Mule 2.2 one can disable temporary jms replyTo destination handling.

Andrew

[Richard Holt]

Hi Andrew,

Thanks for the prompt reply.

Is that on a per endpoint or connector basis?

I need to keep the replyTo working from within the JMSMessageDispatcher but disable it for receiving.

Do you have an example or configuration showing how to disable it?

I will download the latest snapshot and try it.

Regards,

Richard

[Richard Holt]

Andrew,

I have just taken a look at the source code and it doesn't achieve what i need. I need the replyTo for temporary destinations to be as is when sending the data. This is to allow the system to "halt" while i wait for a response from the JMS broker.

It is the reply to response handler itself which i need to disable.

Alternatively am i approaching this problem in the wrong way?

Regards,

Richard

[Andrew Perepelytsya]

You need permanent replyTo destinations, not temporary. Temporary ones are scoped only to a connection, and are visible only to clients using the connection. 2 servers can't share temporary destinations.

Andrew

[Richard Holt]

Sorry Andrew i probably wasn't clear in my response.

I do not want to use Reply To at all, in fact i only use the Temporary Reply To as a result of the remoteSync operation, which is necessary. So my Broker A uses them.

The behaviour i am seeing however is that Mule Broker B wants to use the "JMS Reply To" message header of the inbound message as its Reply To as part of the JMSReplyToHandler.

This means that in the scenario i mentioned originally you either get

A) an error if one of the recipients of the jms message is running slowly (the JMSReplyToHandler of Broker B will error when trying to write to a temporary destination that Broker A has already deleted).

OR

B) Broker A posts to the temporary queue, then listens to it, AMQ Broker A posts the JMS reply and Mule Broker B also posts a message to the temporary queue as well.

Sorry if my explanation's aren't helping grammar is not one of my strong points...

Regards,

Richard

[Andrew Perepelytsya]

Richard,

It sounds like you can just strip the jms replyTo header for the 2nd consumer. Add a MessagePropertiesTransformer on the inbound, search the list archives for reference, it's a pretty typical solution floating around.

HTH,
Andrew

[Richard Holt ()]

Edited as the solution did in fact not work


Hi Andrew,

I looked into the suggestion of removing the property. It appears my initial suggestion for what was wrong was incorrect, it isnt the presence of the header which was driving the creation of the JMSReplyToHandler, this appears to happen as part of the AbstractService as a matter of course.

So as a result of this additional analysis i placed the following into the config i placed above and lo and behold it all worked.

                <service name="testService2">
                  <inbound>
                    <jms:inbound-endpoint queue="test2.queue" connector-ref="localJmsConnector">
                    <response-transformers>
                    <no-action-transformer/>
                    </response-transformers>
                    </jms:inbound-endpoint>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                                <file:outbound-endpoint path="c:/data/out2" connector-ref="fileConnector"/>
                  </pass-through-router>
                  </outbound>
                </service>

Thank you once again for all your help on this issue. I have been working with Mule for about 2 years now and i still keep on finding new things.

Regards,

Richard

[Richard Holt]

Unfortunately even though the test scenario now works it isn't working in my application.

Inbound Transformers run after a ReplyToHandler is set by SEDAService, so i cannot intercept and remove the ReplyTo set by Mule A from within Mule B.

Is there anything which runs prior to the call to construct this ReplyToHandler which i can use to strip the headers?

Regards,

Richard


Here is the configuration again,

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesource.org/schema/mule/core/2.1"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xmlns:spring="http://www.springframework.org/schema/beans"
        xmlns:jms="http://www.mulesource.org/schema/mule/jms/2.1"
        xmlns:jbossts="http://www.mulesource.org/schema/mule/jbossts/2.1"
        xmlns:file="http://www.mulesource.org/schema/mule/file/2.1"
        xmlns:management="http://www.mulesource.org/schema/mule/management/2.1"
        xmlns:context="http://www.springframework.org/schema/context"
        xsi:schemaLocation="http://www.mulesource.org/schema/mule/core/2.1 http://www.mulesource.org/schema/mule/core/2.1/mule.xsd
                http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.1.xsd
                http://www.mulesource.org/schema/mule/jms/2.1 http://www.mulesource.org/schema/mule/jms/2.1/mule-jms.xsd
                http://www.mulesource.org/schema/mule/management/2.1 http://www.mulesource.org/schema/mule/management/2.1/mule-management.xsd
                http://www.mulesource.org/schema/mule/jbossts/2.1 http://www.mulesource.org/schema/mule/jbossts/2.1/mule-jbossts.xsd   
                http://www.mulesource.org/schema/mule/file/2.1 http://www.mulesource.org/schema/mule/file/2.1/mule-file.xsd">
               
        <file:connector name="fileConnector" autoDelete="true" streaming="false"/>

        <jms:activemq-connector name="localJmsConnector" specification="1.1"
                acknowledgementMode="AUTO_ACKNOWLEDGE" recoverJmsConnections="true" persistentDelivery="true">
        </jms:activemq-connector>

        <model name="testModel">
                <service name="testService">
                  <inbound>
                    <file:inbound-endpoint path="c:/data/in" moveToDirectory="c:/save"/>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                    <jms:outbound-endpoint queue="test1.queue" connector-ref="localJmsConnector">
                    <jms:object-to-jmsmessage-transformer/>
                    </jms:outbound-endpoint>
                    </pass-through-router>
                  </outbound>
                </service>

                <service name="testService1">
                  <inbound>
                    <jms:inbound-endpoint queue="test1.queue" connector-ref="localJmsConnector"/>
                  </inbound>
                  <outbound>
                        <custom-outbound-router class="com.acme.JmsRecipientList"/>
                  </outbound>
                </service>

                <service name="testService2">
                  <inbound>
                    <jms:inbound-endpoint queue="test2.queue" connector-ref="localJmsConnector"/>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                                <file:outbound-endpoint path="c:/data/out2" connector-ref="fileConnector"/>
                  </pass-through-router>
                  </outbound>
                </service>

                <service name="testService3">
                  <inbound>
                    <jms:inbound-endpoint queue="test3.queue" connector-ref="localJmsConnector"/>
                  </inbound>
                  <outbound>
                  <pass-through-router>
                                <file:outbound-endpoint path="c:/data/out3" connector-ref="fileConnector"/>
                  </pass-through-router>
                  </outbound>
                </service>
        </model>
</mule>

[antoine.borg]

Hi,

Would the transformFirst attribute on the router help in this case? I'm not
sure if all routers have this attribute though ...

A


Antoine Borg, Senior Consultant | Tel: [phone]
ricston Ltd., BP 2, 1180 Uccle, Brussels, BELGIUM
email: antoine.borg@... | blog: blog.ricston.com | web: ricston.com

-----Original Message-----
From: Richard Holt [mailto:richard_holt@...]
Sent: Wednesday, November 19, 2008 12:24 PM
To: user@...
Subject: Re: [mule-user] JMS Reply To Issue


Unfortunately even though the test scenario now works it isn't working in my
application.

Inbound Transformers run after a ReplyToHandler is set by SEDAService, so i
cannot intercept and remove the ReplyTo set by Mule A from within Mule B.

Is there anything which runs prior to the call to construct this
ReplyToHandler which i can use to strip the headers?

Regards,

Richard
--
View this message in context:
http://www.nabble.com/JMS-Reply-To-Issue-tp20562821p20577852.html
Sent from the Mule - User mailing list archive at Nabble.com.


---------------------------------------------------------------------
To unsubscribe from this list, please visit:

    http://xircles.codehaus.org/manage_email



---------------------------------------------------------------------
To unsubscribe from this list, please visit:

    http://xircles.codehaus.org/manage_email

[Richard Holt]

Hi Antoine,

I have finally realised what i was doing wrong. Essentially i didnt want our system to use remoteSync, instead i needed synchronous. The behaviour i was seeing from the Mule system A and B was perfectly correct.

i.e. for remoteSync

Server A posts to B and awaits a response for 3 seconds. Which is generated by the services in B. I however took to long in B for the response to get back.

In addition i realised that i was more interested in simply posting the message and not put an assumption in place that B is actually present. So instead i post to the AMQ broker sitting between the two but with synchronous=true.

So thanks for everyone's help, it was simply my understanding which wasn't quite correct. Wood for the trees and all that.

Richard