Mailing-List: contact users-help@activemq.apache.org; run by ezmlm
Precedence: bulk
Reply-To: users@activemq.apache.org
Received-SPF: pass (nike.apache.org: domain of gary.tully@gmail.com designates
 209.85.160.171 as permitted sender)
MIME-Version: 1.0
In-Reply-To: 
 <CAH+vQmO=kdjzQDAAExGmHguUDR0Mjm7znsKVoqtDQGJU+zSk8w@mail.gmail.com>
References: 
 <CAMnNrd8VzVDygBuaH2qvEmQzpJMq=_2Ktkco1NNBe_M+1mjeLg@mail.gmail.com>
	<CAH+vQmO=kdjzQDAAExGmHguUDR0Mjm7znsKVoqtDQGJU+zSk8w@mail.gmail.com>
Date: Mon, 12 Mar 2012 23:21:42 +0000
Message-ID: 
 <CAH+vQmPqxdEBSkoMYJw2u3dG-XHCdchBqSgt5GgfP4fF0PLbbg@mail.gmail.com>
Subject: Re: ActiveMQ swallows messages in our environment,
 I need some suggestions
From: Gary Tully <gary.tully@gmail.com>
To: users@activemq.apache.org
Content-Type: text/plain; charset=ISO-8859-1
Content-Transfer-Encoding: quoted-printable

one other thought, assuming non duplex, if you set dispatchAsync=3Dfalse
for your network connector, advisories should always reach the other
broker before messages.

On 12 March 2012 22:21, Gary Tully <gary.tully@gmail.com> wrote:
> I think that is a reasonable theory, depending on load, the order or
> reaction to advisories may not be totally reliable.
>
> One way to validate your theory is to enable the
> sendAdvisoryIfNoConsumers option and subscribe to that advisory to see
> if your messages end up there. The message will be the payload of the
> advisory
> see: http://activemq.apache.org/advisory-message.html
>
> Is your network bridge duplex? I came across an issue[1] where the
> symptom was dropped advisories but it was a problem with unacked
> messages in the duplex case. The workaround is to use two network
> connectors in place of duplex.
> [1] https://issues.apache.org/jira/browse/AMQ-3694
>
>
> On 12 March 2012 21:57, Kai Hackemesser <kai.hackemesser@gmail.com> wrote=
:
>> Hello,
>>
>> - Given a network of V5.5.1 brokers named "brokerA" and "brokerB", and
>> - application "serviceA" is connected to "brokerA" and subscribed to a
>> "request" topic and ready to produce on a "response" topic, and
>> - client application "clientB" is connected to "brokerB", and
>> - named client is first creating a temporary consumer subscription again=
st
>> the "response" topic on "brokerB", then producing a request message for =
the
>> "request" topic, and
>> - "serviceA" received the request message, produced the response message=
 in
>> the response topic on "brokerA",
>> - then the message is not always received by the temporary consumer on
>> broker B (receive times out).
>>
>> This happens often enough to make it an annoyance in our production
>> environment. We have proven that the service responds, but can't underst=
and
>> why the subscriber starves. I currently have the theory that when servic=
eA
>> is responding on brokerA, brokerA is not yet aware of the new
>> consumer/subscriber on the response topic (maybe the network of brokers
>> advisory channels haven't been processed fast enough) and without
>> subscribers the message goes to Nirvana. Can this theoretically be the
>> case? We use Spring JmsTemplate on both ends of the topics and the
>> following class construction to do the synchronous request/response:
>>
>> import java.io.Serializable;
>> import java.util.UUID;
>> import javax.annotation.Resource;
>> import javax.jms.Destination;
>> import javax.jms.JMSException;
>> import javax.jms.Message;
>> import javax.jms.MessageConsumer;
>> import javax.jms.MessageProducer;
>> import javax.jms.ObjectMessage;
>> import javax.jms.Session;
>> import org.springframework.jms.core.JmsTemplate;
>> import org.springframework.jms.core.SessionCallback;
>> import org.springframework.jms.support.JmsUtils;
>> import org.springframework.stereotype.Component;
>>
>> /*
>> =A0* based on
>> http://codedependents.com/2010/03/04/synchronous-request-response-with-a=
ctivemq-and-spring/
>> =A0*/
>> @Component
>> public class SynchronousJmsRequestor {
>>
>> =A0 =A0private static final int TIMEOUT =3D 5000;
>> =A0 =A0private final Destination requestDestination;
>> =A0 =A0private final Destination responseDestination;
>> =A0 =A0private long timeout =3D TIMEOUT;
>>
>> =A0 =A0private final String clientUid;
>>
>> =A0 =A0@Resource
>> =A0 =A0private JmsTemplate jmsTemplate;
>>
>> =A0 =A0public SynchronousJmsRequestor(
>>
>> =A0 =A0 =A0 =A0 =A0 =A0final Destination requestDestination,
>> =A0 =A0 =A0 =A0 =A0 =A0final Destination responseDestination,
>> =A0 =A0 =A0 =A0 =A0 =A0final String clientUid) {
>>
>> =A0 =A0 =A0 =A0this.responseDestination =3D responseDestination;
>> =A0 =A0 =A0 =A0this.requestDestination =3D requestDestination;
>> =A0 =A0 =A0 =A0this.clientUid =3D clientUid;
>> =A0 =A0}
>>
>>
>>
>> =A0 =A0public Serializable request(final String serverUid, final Seriali=
zable
>> payload) throws JMSException {
>>
>> =A0 =A0 =A0 =A0JmsDelegate jmsDelegate =3D new JmsDelegate(payload,
>> requestDestination, responseDestination, serverUid, clientUid,
>> =A0 =A0 =A0 =A0 =A0 =A0timeout);
>>
>> =A0 =A0 =A0 =A0ObjectMessage response =3D (ObjectMessage)
>> jmsTemplate.execute(jmsDelegate, true);
>>
>> =A0 =A0 =A0 =A0return response =3D=3D null ? null : response.getObject()=
;
>> =A0 =A0}
>>
>>
>> =A0 =A0public long getTimeout() {
>> =A0 =A0 =A0 =A0return timeout;
>> =A0 =A0}
>>
>>
>> =A0 =A0public void setTimeout(long timeout) {
>> =A0 =A0 =A0 =A0this.timeout =3D timeout;
>> =A0 =A0}
>> }
>>
>> class JmsDelegate implements SessionCallback<Message> {
>>
>> =A0 =A0private Serializable payload;
>> =A0 =A0private final Destination requestDestination;
>> =A0 =A0private final Destination responseDestination;
>> =A0 =A0private String serverUid;
>> =A0 =A0private final String clientUid;
>> =A0 =A0private long timeout;
>>
>> =A0 =A0public JmsDelegate(Serializable payload, Destination
>> requestDestination, Destination responseDestination,
>> =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 =A0 String serverUid, String cli=
entUid, long timeout) {
>> =A0 =A0 =A0 =A0super();
>> =A0 =A0 =A0 =A0this.payload =3D payload;
>> =A0 =A0 =A0 =A0this.requestDestination =3D requestDestination;
>> =A0 =A0 =A0 =A0this.responseDestination =3D responseDestination;
>> =A0 =A0 =A0 =A0this.serverUid =3D serverUid;
>> =A0 =A0 =A0 =A0this.clientUid =3D clientUid;
>> =A0 =A0 =A0 =A0this.timeout =3D timeout;
>> =A0 =A0}
>>
>> =A0 =A0@Override
>> =A0 =A0public Message doInJms(final Session session) throws JMSException=
 {
>> =A0 =A0 =A0 =A0MessageConsumer consumer =3D null;
>> =A0 =A0 =A0 =A0MessageProducer producer =3D null;
>>
>> =A0 =A0 =A0 =A0try {
>> =A0 =A0 =A0 =A0 =A0 =A0final String correlationId =3D UUID.randomUUID().=
toString();
>>
>> =A0 =A0 =A0 =A0 =A0 * consumer =3D session.createConsumer(responseDestin=
ation,
>> "JMSCorrelationID =3D '" + correlationId + "'");*
>>
>> =A0 =A0 =A0 =A0 =A0 =A0final ObjectMessage message =3D
>> session.createObjectMessage(payload);
>> =A0 =A0 =A0 =A0 =A0 =A0message.setJMSCorrelationID(correlationId);
>> =A0 =A0 =A0 =A0 =A0 =A0message.setStringProperty("CLIENT_ID", clientUid)=
;
>> =A0 =A0 =A0 =A0 =A0 =A0message.setStringProperty("SERVER_ID", serverUid)=
;
>> =A0 =A0 =A0 =A0 =A0 =A0producer =3D session.createProducer(requestDestin=
ation);
>> =A0 =A0 =A0 =A0 =A0 =A0*producer.send(message);*
>>
>> =A0 =A0 =A0 =A0 =A0 =A0*return consumer.receive(timeout);*
>> =A0 =A0 =A0 =A0} finally {
>> =A0 =A0 =A0 =A0 =A0 =A0// Don't forget to close your resources
>> =A0 =A0 =A0 =A0 =A0 =A0JmsUtils.closeMessageConsumer(consumer);
>>
>> =A0 =A0 =A0 =A0 =A0 =A0JmsUtils.closeMessageProducer(producer);
>> =A0 =A0 =A0 =A0}
>> =A0 =A0}
>> }
>>
>> Cheers,
>> Kai
>
>
>
> --
> http://fusesource.com
> http://blog.garytully.com


--=20
http://fusesource.com
http://blog.garytully.com