Expose consumerTag during Consumer_Shutdown event

[plewam]

Hi All,

I subscribe to the Shutdown event of an EventingBasicConsumer. However when I receive the event I cannot track back which consumer was cancelled as I don't have the consumer tag available within the ShutdownEventArgs. This way I cannot evaluate the shutdown reason and eventually reconnect. Would it be possible to add the consumerTag?

I use the a manual ack of rmq messages as a method of throttling if my consumer is not able to process the messages quick enough into a sink (Database in this scenario).
After 30 minutes, rmq cancels my consumer if I have not ack'ed the message. In case of a database maintenance, restart this might happen. In this scenario I would like to reconnect... But I don't know which consumer to reconnect...

On the other hand I also receive the ConsumerCancelled event, but this one only contains the consumerTag and not the shutdown reason.

[michaelklishin]

@plewam are you sure you want to "connect" manually (that is, roll your own connection recovery)?

If you do and realize the amount of effort it will take, please keep in mind that connection recovery is not initiated by consumers. It is initiated by something that reacts to connection shutdown events.

[lukebakken]

@plewam this seems like a reasonable request. I will see if the Java client provides this information (just out of curiosity).

@michaelklishin's point is valid - connection and topolgy recovery is already implemented in this client library. Have you tried it and found it insufficient for your use case?

[plewam]

Alright, your argument is valid. I can do the recover on my own then if I got the consumerTag available.

[lukebakken]

Yes, this library recovers consumers but my guess is that it may not do it in the case of a consumer timeout.

The root cause of this issue is that you are holding on to a message for too long, or taking too long to process it. Can you fix the root cause by modifying your workflow for messages that may take too long to process?

The next fix would be to increase the consumer timeout value. RabbitMQ 3.12 and later allows this to be set per-queue: https://rabbitmq.com/consumers.html#acknowledgement-timeout

You can also set it per-consumer via the x-consumer-timeout consumer argument (code). This somehow did not get documented and I'll get right on that.

[plewam]

Yes it may be because of my implementation. I use rmq as the source of data and a database connector as the sink. If my sink (database) is unable to process the data (e.g. due to the database being down, or updates) I risk flooding my application with data that cannot go anywhere. Therefore I throttle down the ack's of messages. However in my above described scenario I stop ack's completely. The only thing I then can do is to restart the application once every consumer was cancelled. I know kind of try to programm around that. However I got not other idea.

[lukebakken]

Never mind, x-consumer-timeout was removed in rabbitmq/rabbitmq-server#8260

@michaelklishin @deadtrickster - do you remember why the above was removed? I'm just curious.

[michaelklishin]

We didn't see the value in having not just per queue but per consumer values. How common is it for N consumers on the same queue to have different processing timeouts?

[lukebakken]

Hello @plewam -

I will have a new alpha release for you to test soon.

With regard to EventingBasicConsumer, when the Shutdown event is raised, the sender argument will be the instance of the EventingBasicConsumer that is raising the event, and this instance will have a ConsumerTags[] property where you can find the consumer tags for the event.

I'm going to close this discussion, but let me know if this meets your requirements.

[plewam]

@lukebakken Perfect, thats exactly what I need. Thank you