RabbitMQ Client .Net Core Timing Out under load after rabbitmq upgrade. Consumer System.TimeoutException

[KaizerT]

We have a system that was running RabbitMQ 3.8.5 with RabbitMQ.Client 5.2.0. Part of the way the system uses rabbit is by creating a consumer to wait for a specific command from a different executable when a long-polling web request comes in from a client. The web request comes in on a .Net 6.0 web APIA consumer is created per client since said command is posted on a queue specific to that client. After a given timeout, the consumer is disposed and the client restarts the long-polling request. We were able to support 1500+ clients/consumers simulataneous long-polling requests. This worked fine until we upgraded to RabbitMQ 3.13.3 with RabbitMQ.Client 6.8.1. A timeout exception now occurs when about 700 clients initiate this long-polling request. The same happens when 500 clients are connected for a long amount of time (> 1 hour). We have also tried using EasyNetQ, since we use the library in other parts of the system, for consumption and got the same error and behavior, a timeout followed by the connection being terminated by the RabbitMQ Server.

Further investigation seems to indicate that the hearbeats are missed when the timeout exception occurs, and the RabbitMQ server terminates the connection, causing a domino effect of failures on our side. We have yet to figure out what's causing the timeout to occur and block all operations related to RabbitMQ.

Some details are providede below:

Here is the initial exception that occurs when the error first occurs:

    at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
    at RabbitMQ.Client.Impl.ModelBase.QueueDeclare(String queue, Boolean passive, Boolean 
     durable, Boolean exclusive, Boolean autoDelete, IDictionary`2 arguments)
    at RabbitMQ.Client.Impl.ModelBase.ConsumerCount(String queue)

EasyNetQ error

System.TimeoutException: The operation has timed out.
   at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
   at RabbitMQ.Client.Impl.ModelBase.QueueDeclare(String queue, Boolean passive, Boolean durable, Boolean exclusive, Boolean autoDelete, IDictionary`2 arguments)
   at RabbitMQ.Client.Impl.AutorecoveringModel.QueueDeclare(String queue, Boolean durable, Boolean exclusive, Boolean autoDelete, IDictionary`2 arguments)
   at EasyNetQ.RabbitAdvancedBus.<>c__DisplayClass51_0.<QueueDeclareAsync>b__0(IModel x)
   at EasyNetQ.Persistent.PersistentChannel.InvokeChannelActionAsync[TResult,TChannelAction](TChannelAction channelAction, CancellationToken cancellationToken)
   at EasyNetQ.RabbitAdvancedBus.QueueDeclareAsync(String name, Action`1 configure, CancellationToken cancellationToken)
   at EasyNetQ.AdvancedBusExtensions.QueueDeclare(IAdvancedBus bus, String name, Boolean durable, Boolean exclusive, Boolean autoDelete, CancellationToken cancellationToken)

Initial Consumer Code

 ncChannel.QueueDeclare(queueName, true, false, false, null);
 ncChannel.ExchangeDeclare(ExchangeName, ExchangeType.Topic, true, false, null);
 ncChannel.QueueBind(queueName, ExchangeName, routingKey);
Channel<BasicDeliverEventArgs> responseChannel = Channel.CreateBounded<BasicDeliverEventArgs>(channelOptions);
EventHandler<BasicDeliverEventArgs> nextCommandHandler = async (object sender, BasicDeliverEventArgs msg) =>
{
    try
    {
        if (msg != null && msg.Body.ToArray() != null)
        {
            string msgBody = Encoding.UTF8.GetString(msg.Body.ToArray());
            //log
        }
        else
        {
            //log
        }
        await responseChannel.Writer.WriteAsync(msg);
        responseChannel.Writer.TryComplete();
    }
    catch (Exception ex)
    {
        //log
    }
};
consumer.Received += nextCommandHandler;
 bool autoAck = true;
 consumerTag = ncChannel.BasicConsume(queueName, autoAck, consumer);

Tried changing the consumer to an async eventing basic consumer

Channel<string> responseChannel = Channel.CreateBounded<string>(channelOptions);

ncChannel.QueueDeclare(queueName, true, false, false, null);
ncChannel.ExchangeDeclare(ExchangeName, ExchangeType.Topic, true, false, null);
ncChannel.QueueBind(queueName, ExchangeName, routingKey);
AsyncEventingBasicConsumer consumer = new AsyncEventingBasicConsumer(ncChannel);
AsyncEventHandler<BasicDeliverEventArgs> nextCommandHandler = async (object sender, BasicDeliverEventArgs msg) =>
{
    try
    {
        if (msg != null && msg.Body.ToArray() != null)
        {
            string msgBody = Encoding.UTF8.GetString(msg.Body.ToArray());
            //log
        }
        else
        {
            //log
        }
        await responseChannel.Writer.WriteAsync(msg);
        responseChannel.Writer.TryComplete();
        await Task.Yield();
    }
    catch (Exception ex)
    {
        //log
    }
};
consumer.Received += nextCommandHandler;

bool autoAck = true;
consumerTag = ncChannel.BasicConsume(queueName, autoAck, consumer);

EasyNetQ Consumer Code

var queue = _rabbitBus.Advanced.QueueDeclare(queueName, true, false, false);
consumer = _rabbitBus.Advanced.Consume(queue, (body, properties, info) => Task.Factory.StartNew(async () =>
{
    try
    {
        string msgBody = null;
        if (body.ToArray() != null)
        {
            msgBody = Encoding.UTF8.GetString(body.ToArray());
            //log
        }
        else
        {
            //log
        }
        await responseChannel.Writer.WriteAsync(msgBody);
        responseChannel.Writer.TryComplete();
        await Task.Yield();
    }
    catch (Exception ex)
    {
        //log
    }
}));

[lukebakken]

Hello,

Other than upgrading RabbitMQ and the version of this library, what else changed?

• Workload?
• Server specs?
• Operating system versions?
• Did you change from classic to quorum queues?

Further investigation seems to indicate that the hearbeats are missed when the timeout exception occurs,

Is there evidence for this? Having RabbitMQ logs would be very helpful. Please point out the specific time(s) that these timeouts occur, so I can correlate it in the logs.

Finally, if you can provide code to reproduce this issue, or at least provide a much more complete example of what you are doing, it would help. Right now you're asking for a lot of guessing on my part which I don't have time to do.

And, in addition, I strongly recommend trying the latest RC for version 7 of this library:
https://www.nuget.org/packages/RabbitMQ.Client/7.0.0-rc.5

Testing it would help us out greatly, plus there's a chance it will address your issue.

[KaizerT]

Hello Luke,

I did try with RC-9 and it seems to no longer encounter the issue, but I only converted the test app and not our app since it'd be too complex.

[KaizerT]

Just to update this discussion, we're still encountering the issues with our main app, I was in the middle of making a test run for RC-14 on our main app since we couldn't figure out a way to do this, when we noticed the release for 7.0.0. I converted all I could with minimal changes to our code, but since there were a lot of interdependencies I still needed to update a lot of the code. Performance was significantly worse after I updated to 7.0.0, but not we're not really sure if it was a direct result of upgrading to 7.0.0, or if any of the changes we made to accommodate the migration might have caused it. We've decided to try to get 6.8.1 to perform, since it's late in our release cycle. We'll try again with 7.0.0 on our next release since we also have EasyNetQ on our project which hasn't updated to 7.0.0

[lukebakken]

Please see this discussion:

#1721

My guess is that EasyNetQ assumes that IChannel is not thread-safe the same way that MassTransit does (or, did).

[KaizerT]

Looking through their commits, it was just cuz they(EasyNetQ) haven't done any changes related to rabbitmq-client 7.0.0. For our test we just removed all the easynetq references since the problem area is using rabbitmq client directly. I'll check it out though.

[michaelklishin]

@KaizerT to clarify, after switching to using the client directly without the (not yet updated version of) EasyNetQ, you did observe a significant behavior change?