JMS QueueBrowser question

[ftg314159]

I have a multithreaded application where one thread writes messages to a queue, and another thread accesses these messages via a QueueBrowser.  The thread run simultaneously.

The second thread wants JMS to hold onto the messages until it is done with individual messages, so it uses QueueBrowser.getEnumeration() to get an enumeration, and then steps through the messages.  Each message is added to a HashMap along with a Boolean that indicates whether the app is done with it (has logically released it).

Because the JMS spec says that acknowledging any message automatically acknowledges all messages received previous to the acknowledged messages, the app can't directly receive and acknowledge messages in order, as it may not be done with some of them.  Whenever we update the HashMap, we look through the message entries in order starting with the oldest, and do a receive/acknowledge for each logically released message until we find one which has not been logically released.  Basically, we only receive/acknowledge a message if we know that we are done with all the previous messages.

Obviously, this relies on the enumeration returned by QueueBrowser being accurate.  But I get repeated results where the QueueBrowser enumerations do not include messages which have been sent to the queue.  In each case, one or more messages are missing from the start of the set (i. e. the earliest messages).  It is never the case that the latest messages are missing, so this can't be a case where the first thread's send() hadn't completed at the time the enumeration was constructed.

The release logic checks to see whether receive() gives it the JMS message ID it expects, based on the value saved in the HashMap entry, and I'm finding that the messages that were excluded from the enumeration show up in the proper order when receive() is issued.

If I run the sending thread by itself, and then run the receiving thread by itself when the sending thread has ended, all works as expected and no messages are lost.

I've verified this problem with both EAP 6.3 and an old pre-HornetQ JBossAS 5.1.0.

Am I misunderstanding how QueueBrowser is supposed to work ?

[ataylor]

I'm a bit confused with your post, you say your using a Queue Browser but talk about acknowledging messages which makes no sense, unless you mean you are using a Queue browser and a normal receiver on the same queue.

If the latter then bear in mind that the queue browser will only return messages still in the queue and not those out for delivery, that is, in the consumers buffer.

[ftg314159]

I am using a QueueBrowser and a normal receiver on the same queue.  The point of the design is to access the messages in order for the application but release them in a random order.  The messages contain event traces of transaction sessions, and we don't want to release/acknowledge a message that contains a BeginTransaction unless we've encountered the EndTransaction, which may only show up several messages down the line.  The normal receiver is used *only* to release/acknowledge messages that we know we're done with.  The application accesses the message content *only* through the QueueBrowser.

When you say "out for delievery", do you mean messages received in CLIENT_ACKNOWLEGE mode but not yet acknowledged, or are you saying that the Connection/Session just arbitrarily removes a bunch of the earliest messages from the QueueBrowser's view because it thinks you may receive them later ?  Will Connection.stop() influence this behavior ?

[ftg314159]

Actually, I just checked the code, and we only call Connection.start()/stop() around the code that actually receives/acknowledges messages.  The Connection is always stopped when getEnumeration() is called.

Can you point me to a discussion of the "out for delivery" concept ?  The JMS API Javadoc doesn't seem to mention this, and only says that messages can arrive at the queue or be received/acknowledged while the enumeration exists.  I have no problem not seeing messages which arrived after the enumeration was constructed, as I'll see them the next time, and nothing is receiving/acknowledging messages when getEnumeration() is done.

[ataylor]

Actually, I just checked the code, and we only call Connection.start()/stop() around the code that actually receives/acknowledges messages.  The Connection is always stopped when getEnumeration() is called.

stopping and restarting the connection will make no difference, see next comment.

Can you point me to a discussion of the "out for delivery" concept ?  The JMS API Javadoc doesn't seem to mention this, and only says that messages can arrive at the queue or be received/acknowledged while the enumeration exists.  I have no problem not seeing messages which arrived after the enumeration was constructed, as I'll see them the next time, and nothing is receiving/acknowledging messages when getEnumeration() is done.

basically when i say 'out for delivery' i mean that the message has been removed from the queue and sent to the client. A consumer will buffer messages ready for receive() to be called. this is how most messaging vendors do this mostly because of performance reasons. Its also the reason the spec is vague about what a queue browser returns.

to summarize, a browser will return the messages i a queue at any point in time, any messages currently 'out for delivery' will not be there.

I'll be honest, reading between the lines of your use case, Im not sure using JMS is the best approach, you mentioned receiving messages out of order which sort of goes against queueing, altho i am making some assumptions. maybe you could decouple the 2 parts of your business logic

[ftg314159]

Based on your last post, my next test was going to be not instantiating the Consumer until I was ready to receive/release/acknowledge messages, and then closing it, figuring that that may prevent the type of advance buffering you describe.

Frankly, I find the design you're describing a little strange.  It smacks of something some messaging vendors were doing pre-JMS and refused to change.  The JMS contract requires the server to retain knowledge of unacknowledged messages no matter what, so I don't see what purpose is served by excluding buffered messages from a browse; the server has to know about them anyway, and until they're actually consumed they should be fair game for any potential recipient.

I didn't want to muddy the waters of the JMS question with too much of the business logic, but here's what's going on.  A transaction processor generates a stream of event traces which describe the changes made by the transactions.  These event traces are collected into fixed-length blocks, written to a round-robin set of collection files, and can be used (a la the CICS journal) to reapply the changes to a backup system later.  For this reason, they are copied to archive datasets whenever the system switches between journal files;  the one just switched-from is copied, and becomes available for reuse.  They can also be captured in realtime to apply the same changes to a duplicate system, which is what we do.  Each time a journal is copied, we update a database with the archive filename and the range of records it contains.

The realtime capture is catch-as-catch-can.  It can get interrupted because the JMS client writing the data to a queue can't keep up, or the client may be down, etc.  If the client determines that a gap exists, i. e. that some records are missing, it notifies the capture code which consults the database, determines which archived copies contain those records, retrieves them, and sends them down to the client.

The client uses two Queues, a Primary and an Auxiliary.  Normally, records are written to the Primary.  When a gap is detected, records get diverted to the Auxiliary until the missing records have been retrieved and written to the Primary, at which point the contents of the Auxiliary are transferred to the Primary and normal capture resumes.

The Primary is read by a component that applies the changes to a duplicate system, and relies on the fact that records don't make it to the Primary until any gaps are remediated.  The point of all this is to leverage the JMS contract that guarantees to retain persistent messages in order to avoid having to do that ourselves.

[ataylor]

Based on your last post, my next test was going to be not instantiating the Consumer until I was ready to receive/release/acknowledge messages, and then closing it, figuring that that may prevent the type of advance buffering you describe.

yes it would, but could be slow.

Frankly, I find the design you're describing a little strange.  It smacks of something some messaging vendors were doing pre-JMS and refused to change.

I'm not sure why you would say that, HornetQ is post JMS and of the several messaging brokers Ive developed on  most of them do the same, the performant ones anyway.

The JMS contract requires the server to retain knowledge of unacknowledged messages no matter what, so I don't see what purpose is served by excluding buffered messages from a browse; the server has to know about them anyway, and until they're actually consumed they should be fair game for any potential recipient.

Your partially correct, the server has to know about them for acknowledgment purposes but it doesn't need the message to exist in the queue and as long as ordering rules are maintained messages can be routed to consumers in bulk. its the only way to get fast throughput.  I will reiterate what I said about queue browsers, the spec is vague for a reason, it doesnt actually specify what should be returned and this is because to much is dependent on the Vendors implementation.

I didn't want to muddy the waters of the JMS question with too much of the business logic, but here's what's going on.  A transaction processor generates a stream of event traces which describe the changes made by the transactions.  These event traces are collected into fixed-length blocks, written to a round-robin set of collection files, and can be used (a la the CICS journal) to reapply the changes to a backup system later.  For this reason, they are copied to archive datasets whenever the system switches between journal files;  the one just switched-from is copied, and becomes available for reuse.  They can also be captured in realtime to apply the same changes to a duplicate system, which is what we do.  Each time a journal is copied, we update a database with the archive filename and the range of records it contains.

The realtime capture is catch-as-catch-can.  It can get interrupted because the JMS client writing the data to a queue can't keep up, or the client may be down, etc.  If the client determines that a gap exists, i. e. that some records are missing, it notifies the capture code which consults the database, determines which archived copies contain those records, retrieves them, and sends them down to the client.

The client uses two Queues, a Primary and an Auxiliary.  Normally, records are written to the Primary.  When a gap is detected, records get diverted to the Auxiliary until the missing records have been retrieved and written to the Primary, at which point the contents of the Auxiliary are transferred to the Primary and normal capture resumes.

The Primary is read by a component that applies the changes to a duplicate system, and relies on the fact that records don't make it to the Primary until any gaps are remediated.  The point of all this is to leverage the JMS contract that guarantees to retain persistent messages in order to avoid having to do that ourselves.

Have you though about an MDB sitting in the middle and deciding if the primary queue gets the message? have a single queue where all messages go, these are consumed by the MDB who decides whether to fwd them to either the primary queue or auxiliary, if the MDB steps into auxiliary mode, it waits for the missing messages and then moves the ones it needs to.

I'm on pto until next week now, but I will gladly help you some more if I can. Or catch me on IRC next week.

[ataylor]

another solution would be to have 2 queues on the same address (or topic with 2 subscriptions), each queue gets all the messages and each consumer consumes from its own queue, this way they wont effect each other.

[clebert.suconic]

Frank.. you could have made it easier by going straight to the point here.. I still don't understand your architecture.. and I'm not sure it matters to the problem.

It sounds a bit weird that you are actually using Browser to do any processing.... once the message is acknowledged.. the message is gone... Browser is meant to browse messages on a tool where you don't actually care about receiving it. If another thread acks the message before you get to the message you won't see it.

For cases like this you really need a topic with two subscriptions, one for each thread...     

Again.. I don't fully understand your logic.. and it's being a bit hard for me to figure out here.

[ftg314159]

Your partially correct, the server has to know about them for acknowledgment purposes but it doesn't need the message to exist in the queue and as long as ordering rules are maintained messages can be routed to consumers in bulk. its the only way to get fast throughput.  I will reiterate what I said about queue browsers, the spec is vague for a reason, it doesnt actually specify what should be returned and this is because to much is dependent on the Vendors implementation.

I'm not suggesting that the server shouldn't pre-load a Consumer's queue for performance reasons with messages it may or may not consume.  I'm just saying that since the server must retain those messages until they are acknowledged there is no reason I can fathom for them to be left out of a browser enumeration.  What does it matter that they have been sent anticipatorily to a client for performance reasons ?  If another client for the same queue came along and did a receive, he would get the next unreceived message whether or not that message had been farmed out "on spec" to other clients who had not yet received it, would he not ?

I took your point about the spec being vague.  And I wasn't poking at HornetQ in particular; as I said, I got the same results using JBoss 5.1.0 which used its own JMS.  What I meant was that the JMS spec is probably deliberately vague because some pre-JMS vendor didn't want to modify his design to accommodate the browser behavior I'm wishing for.

Have you though about an MDB sitting in the middle and deciding if the primary queue gets the message? have a single queue where all messages go, these are consumed by the MDB who decides whether to fwd them to either the primary queue or auxiliary, if the MDB steps into auxiliary mode, it waits for the missing messages and then moves the ones it needs to.

That's essentially what's happening now, except no MDB is used.  Blocks coming in are sent to Primary as long as they are the next expected block.  When a gap is recognized, blocks go to Auxiliary unless they happen to be the next expected block for Primary, in which case the block is written to Primary and the gap shrinks by one block.  When the first block in Auxiliary becomes the next expected block for the last block written to Primary, the gap is closed, and Auxiliary is transferred to Primary (at least up to the point where another gap is detected in the stuff Auxiliary has accumulated)

[ftg314159]

another solution would be to have 2 queues on the same address (or topic with 2 subscriptions), each queue gets all the messages and each consumer consumes from its own queue, this way they wont effect each other.

I think you've got the wrong end of the stick here.  There is only one Consumer.  There are two Threads, one for a Producer which collects blocks from an external source and sends them to either the Primary or Auxiliary Queues, and one for a Consumer who consumes only the Primary Queue (and knows nothing about the Auxiliary).

[ftg314159]

Frank.. you could have made it easier by going straight to the point here.. I still don't understand your architecture.. and I'm not sure it matters to the problem.

It doesn't, which is why I didn't include it in the initial post.  I added it later because Andy was curious about why I was doing what I was doing.

It sounds a bit weird that you are actually using Browser to do any processing.... once the message is acknowledged.. the message is gone... Browser is meant to browse messages on a tool where you don't actually care about receiving it. If another thread acks the message before you get to the message you won't see it.

Yes, I'm well aware of that.  There is only ever one Thread using a Consumer for the Primary Queue.  The point of the browser is that the app may process messages 1, 2, 3, 4, 5, and want to release 2, 3, and 4, but not 1.  JMS receive/acknowledge doesn't allow that.  As soon as you call acknowledge(), all 5 are gone.  The browser allows the Consumer thread to access and deliver 1-5 to its app without relieving JMS of the need to retain those messages.  If stuff crashes, we are guaranteed that 1-5 will be redelivered.  At some later point, the app may have now seen 6, 7, and 8, and it may be done with 1-5 but not 6-8, so we now do 5 receive/acknowledge calls which receive and acknowledge 1-5.  6-8 are not affected, since they've only been accessed through the browser.

[ftg314159]

OK, with Andy's help about the "out for delivery" concept, I found a solution.  The reality is that you can't have the QueueBrowser and the Consumer in existence at the same time.  The QueueBrowser has to be created and closed around getting the enumeration, and the Consumer has to be created and closed around doing the receive/acknowledge calls.  If I do that, all works as expected.

Not only does closing the Consumer before creating the QueueBrowser prevent the QueueBrowser from missing messages that were sent to the Consumer's buffers, but not closing the QueueBrowser before activating the Consumer results in the Consumer missing messages that might be in the QueueBrowser's buffers.

Personally, I think this is nuts, and I'll open a JIRA about it.  Andy, thanks again for your help.