No Consumer Buffering caused random Queue behavior

[riteshj2ee]

Hi All,

Earlier when new Consumer starts, it wasn’t able to pick-up the messages from Queue even though messages are there in Queue because

messages were internally buffered in Consumers which are already running and that’s is the default behavior of HornetQ.

This problem was resolved by introducing No Consumer Buffering in Queue, But after that Queue behavior became unstable.

Eg. After picking up the few messages from Queue, sometimes Consumers are not picking-up the next available messages.

Now to resolve this problem, I am just closing the Session on existing Connection after receiving one message from Queue.

So Connection will be established to Queue only once but every time new Session will be created before receiving a message and

Session will be closed after receiving a message from Queue. 

Can anybody tell me is this approach is fine or not?

Earlier without introducing No Consumer Buffering, single Session was working as per expectation, so why its not working fine with

No Consumer Buffering?

CODE: I am using JBoss 6AS HornetQ with JMS on Linux OS

// initialization code
queueConnectionFactory = (QueueConnectionFactory) jndiContext.lookup(QUEUE_CONNECTION_FACTORY);
queue = (Queue) jndiContext.lookup(JNDI_QUEUE_NAME);
queueConnection = queueConnectionFactory.createQueueConnection();
queueConnection.start();
// code in loop
queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);
Message m = queueReceiver.receive();
queueSession.close();

Thanks & Regards,

Ritesh

[clebert.suconic]

Earlier when new Consumer starts, it wasn’t able to pick-up the messages from Queue even though messages are there in Queue because

messages were internally buffered in Consumers which are already running and that’s is the default behavior of HornetQ.

This problem was resolved by introducing No Consumer Buffering in Queue, But after that Queue behavior became unstable.

Eg. After picking up the few messages from Queue, sometimes Consumers are not picking-up the next available messages.

That's the same with ***every*** serious message system I know, including Tibco, ActiveMQ, JBoss Messaging, WebSphere MQ and HornetQ.

Buffering is a common feature towards performance and avoid network latency.

Now to resolve this problem

There's no problem here.. you just have to consumer your message accordingly.

Perhaps you are leaking consumers?

If you are doing something like this:

for (int i = 0 ; i < 1000; i++)

{

     queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

     queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);

     Message m = queueReceiver.receive();

}

It means you need to research a bit more about Message Systems...  You are creating a consumer that will poll messages out of your queue and you would be leaking it.

I mean no offense, but that would be something really, really dumb!

[clebert.suconic]

BTW: you are using the wrong forum, please use the user's forum.

I'm now moving your post to the right forum.

[riteshj2ee]

Hi Clebert,

Thanks for reply, I was using below code earlier :

(CODE BLOCK 1) :

queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);

for (int i = 0 ; i < 1000; i++) {

     Message m = queueReceiver.receive();

     // process message

}

But after introducing NO CONSUMER BUFFERING, above code was started behaving randomly, It was not receving messages

from Queue even though messages were there in Queue, So I tried below code with that I am getting consistent behavior:

(CODE BLOCK 2) :

for (int i = 0 ; i < 1000; i++) {

     queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

     queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);

     Message m = queueReceiver.receive();

     // process message

}

I know above code looks really really dumb, but thats because of really really dumb behavior from Queue.

Try to understood my queustion properly that : With CONSUMER BUFFERING, CODE BLOCK 1 works fine but it won't work with NO CONSUMER BUFFERING, So I tried with CODE BLOCK2 and its working fine now, so my question is Why CODE BLOCK1 is not working fine with NO CONSUMER BUFFERING ?

I've put this discussion in both kind of forums and in any one of the forum I've attached my complete code, and it won't take more than 5 minutes

to reproduce this problem.

I am available anytime on google talk: ritesh.j2ee

Thanks.

-Ritesh

[clebert.suconic]

(CODE BLOCK 2) :

for (int i = 0 ; i < 1000; i++) {

     queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

     queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);

     Message m = queueReceiver.receive();

     // process message

}

^^ Really, this is the worse code I have ever seen ;-)

I can't talk to you on google talk now...

Look at any JMS tutorial. At least don't create 1000 consumers. (That's what you are doing now).. The system is sending the load balancing messages with each one of these 1000 consumers.

You will have at least one message buffering on each consumer at some point.

Just fix your code!

[riteshj2ee]

Sorry I've not put my complete code, its actually like this:

(CODE BLOCK 2) :

for (int i = 0 ; i < 1000; i++) {

     queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);

     queueReceiver = queueSession.createReceiver(queue, receiverMessageSelector);

     Message m = queueReceiver.receive();

     // closing the session after receiving the message

     queueSession.close();

     // process message

}

So I think it won't create 1000 Consumers on single Connection object, there will be one Consumer/Session over single Connection but

keeps Closing & Creating every time for every message received.

[clebert.suconic]

That's even worse!

Just create a single Consumer, Message Consumer... whatever!

Or create a MDB!

[clebert.suconic]

Use a message listener for instance.

[riteshj2ee]

I can't use Message Listener because I need to fetch/receive the next message only after the completion of message processing of received message.

[clebert.suconic]

You really need to get some basic information on JMS... Read a tutorial / docs / books... etc

you just use a message Lisnster

public MyListener ....

{

    public void onMessage(Message msg)

   {

            /// do your work here

            /// you won't receive another message until you finished the first one. how that would be any different?

   }

}

[riteshj2ee]

The problem is reproducible in hornetQ 2.2.5 as well.

I've attached complete code with this.

Please do needful.

Thanks.