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
    }
}));