Draining pre-fetched messages from consumer c#

[dazinator]

I have a pool of channels in my dotnet application, load balanced over several connections.
I'd like to create multiple consumers, and I am using AsyncEventingBasicConsumer for that, each AsyncEventingBasicConsumer has its own channel leased from the channel pool.
I'm handling the Received event.
I understand there is a hidden "cache" of prefetched messages - for example 10 and that the Received event is fired for one message at a time but all 10 may be received in one go as far as rabbitmq is concerned all 10 are awaiting acknowledgement, until each one is the then processed by the Receive handler on the consumer and Ack'd / Nack'd.

Here is where my issues start (not surpisingly with additional complexity)
In order to take more control over concurrency concerns (in terms of how my application processes messages), in the Received event handler, I'm pushing the deserialized message (making sure to grab the byte[]'s whilst they are available within the handler as docs suggest) into my own System.Threading.Channel which acts as my own buffer. I'm going to refer to this as a System.Threading.Channel as the term channel already refers to a rabbit mq channel. In this case I'm using an unbounded (no max limit on size), single writer, single reader System.Threading.Channel- however in future scenarios I am planning to use a single writer, multiple reader System.Threading.Channel - where the multiple readers in my application would compete for messages from the System.Threading.Channel. The latter will be for scenarios where I care about processing throughput more than preserving an ordered processing sequence.
When the AsyncEventingBasicConsumer receives the message, within the Received event handler, I push my deserialized representation of the received message into the System.Threading.Channel along with:-

• The message deliveryTag
• A couple of callbacks that can be invoked out of band of this event handler, to perform either a BasicAck or BasicNack of this message on the channel later on.

The reader of the System.Threading.Channel then reads messages asynchronously, and invokes the BasicAck or BasicNack callbacks depending upon if it processed the message successfully.

My first questions it that I am seeing something funny behavior when the reader calls BasicNack (or BasicReject) in this way. It appears to me that the messages does not return to the queue after calling BasicNack. I am not using any prefetch count qos setting so the behaviour I see with 3 messages ready on a queue, is that all 3 enter the state ready together (i.e they are all prefetched together which I understand due to having not set a prefetch count all messages are sent together), then upon processing the first message from the System.Threading.Channel the reader calls BasicNack on that message (at this point there is also another message in the channel) but in rabbitmq after the BasicNack is complete for this first message (also tried BasicReject) all 3 messages all still show as waiting acknowledgement.

The messages do all return to the queue (ready state) if I close the channel..
However given that I'd like to return the channel to the channel pool I'd rather not have to close the channel, I'd rather keep the channel open for re-use.
So my first question is: I am wondering if there is any reason that might explain the the behaviour I think I am witnessing, in terms of BasicReject / BasicNack not working when called later on for message (out of band from the Receive event handler).

My next question is around re-use of channels amongst subscribers.
Assuming the single reader of the System.Threading.Channel hits a problem and can't process a message, and BasicNack`s it. It's use case is that if it can't process a message it stops there and will not process any more - as it cares about preserving order when processing. The idea is the issue with the current message must be resolved before processing would continue. It might be that it eventually gets moved to a dead letter queue for example and thats ok. However at the point the reader decides to "stop" at a message that cannot be processed, there might still be:-

• Another 5 messages sitting in the prefetch cache which the AsyncEventingBasicConsumer has not fired Received for yet.
• Another 4 messages that the AsyncEventingBasicConsumer has fired the Received event for, and were subsequently pushed to the System.Threading.Channel and are waiting there behind the current message for the reader to pick them up still - but it doesn't want to continue to process them as it wants to "stop" at the current message.

In this situation in order to keep the channel alive, but "free up" the messages so they are available for redelivery to the next consumer that is created on the same channel (to replace the current consumer that has experienced the issue that may be transient), I think I need to do the following:

1. Stop the consumer from receiving any more messages

• I call currently call channel.BasicCancel(consumer) - hoping this is correct

2. Send a Nack for any messages sitting in the pre-fetch cache

• Not sure how I do this or access these?

3. Send a Nack for any messages sitting behind the current one (waiting to be processed) that my reader is processing from the System.Threading.Channel.
4. If any of this goes wrong, close the channel - closed channels aren't returned to the pool, instead a new channel will be created as needed within the pool balanced accross one of the open connections.

I believe I can do 1, 3, and 4 but not sure how to go about 2, or if there is a better way to handle this scenario.

[lukebakken]

but in rabbitmq after the BasicNack is complete for this first message (also tried BasicReject) all 3 messages all still show as waiting acknowledgement.

RabbitMQ has already re-delivered the message to your consumer at this point.

If I understand your second scenario where you have in-flight deliveries for your client to process but your client experiences an error, the simplest action is to close the channel. You don't even have to cancel the consumer as the .NET client will clean up consumer instances when the IModel is disposed. As soon as RabbitMQ knows the channel is closed any consumers will be cancelled and all outstanding messages will be re-enqueued and delivered to the next available consumer of that queue.

If you have further questions the best way to hash things out would be to provide some code I can clone, compile and run to see the behavior. Even better would be to make the repository available via a git hosting provider that allows pull requests so that some collaboration is possible.

[dazinator]

RabbitMQ has already re-delivered the message to your consumer at this point

Ah that makes a great deal of sense. I'll try a verify that, thank you.

the simplest action is to close the channel.

Ok. I think I'm possibly trying to over optimise and complicate things by attempting not to close the channel and instead reuse it. I know that channel creation has overhead hence why I created a channel pool concept - which I use heavily on the publishing side. On the consumer side a channel pool seems less useful at the moment - when you are listening indefinitely for messages - consuming a channel indefinitely means it's only returned to the pool on some error, and now we are saying actually - on error you need to close the channel anyway so it can't even be returned to the pool in that scenario. The only scenario a channel pool remains useful on the subscriber side is if you need to listen to a finite number of messages and expect to process those cleanly then return the channel to the pool for the next thread to possibly do the same. Having a client subscribe to a finite number of messages isn't a use case I have encountered yet but possibly will in the future. I guess I'll keep it simple for now and close the channel which will remove it from the pool.

[lukebakken]

Yep, check the redelivered flag on the message and it should be true.

[dazinator]

Yep just confirming this is what was happening. I was nack'ing the messages and they were relivered so quickly that I thought they weren't being nack'd :-) - I wasn't seeing them because they were sitting back in the pre-fetch cache again at the point I was deciding to halt processing and inspect rabbit mq.

As I haven't figured out a way to properly nack all unacknowledged messages for a consumer, without closing the channel (whilst also preventing re-delivery) i'm just going to resort to closing the channel for now as explained below in another comment.

Thanks for helping me figure this one out though!

[michaelklishin]

If you need to requeue all delivered messages outstanding an ack, simply close the channel they were delivered on. You can check if a delivery is a repeated one from delivery metadata.

With quorum queues, you can limit the maximum number of attempted redeliveries, too.

[dazinator]

Thanks. Yes I think I'll try not to reuse the channel and simply close it for now. I'll also take a look at that metadata. I am indeed using quorum queues.

[michaelklishin]

When dynamically using channels like this, beware of channel leaks and churn.

[dazinator]

Thanks. This was originally why I would like to re-use the channel without closing it (to mirror what I am doing on the publishing side without any issues). However its difficult to manage nack'ing the messages that may be in the prefetch cache whilst also preventing re-delivery, so this makes doing this infeasible, meaning my only viable option at present is to close the channel unless I can find a solution. However given the simplicity of closing the channel, and given the fact this would only be done on exception (the normal status quo of the system is the messages are processed successfully) I think this is ok. I'll manage channel churn in this scenario by putting in a delay between closing the channel, and creating a new one, to give any time for a transient error to clear. In combination with dead lettering problematic messages, I think this should be enough to ensure any channel churn is temporary!

[michaelklishin]

In this case closing the channel and increasing the churn sounds OK. Many beginners open a new channel for every message published, which ruins publishing throughput and results in extremely high channel churn, wasting a lot of resources.