Create model timeout after consumer ack timeout

[pabermod]

Hi, i'm testing the following scenario:

I have a consumer that can take more time than the configured consumer_timeout.

What I want is when the channel is closed due to this case, create a new channel and a new consumer to the same queue (the first consumer won't receive more messages than the prefetched ones).

If the message I receive in the consumer has been resend, I won't process it, I will only ack it.

But when I try to create another channel I got a timeout

consumer_timeout is set to 1000

Also I set channel_tick_interval to 500 in advanced.config

I attach a receive client demo ReceiveRabbit.zip

[pabermod]

@michaelklishin @lukebakken Is this the expected beahivour? I don't see why i cannot open a new channel after it is closed due to an ack timeout

[lukebakken]

consumer_timeout is set to 1000

Is that message from this client or from RabbitMQ? What is logged by RabbitMQ when you try to create a new channel?

I am very busy with support for paying customers but I'll check this out eventually. I appreciate you providing code!

[pabermod]

Hi, thanks for your time, after doing some testing, the timeout while recreating the channel does not occur always:

As you can see in the server does not log anything when the exception.

The code to reopen the channel is the following:

[pabermod]

Hi, I have another doubt regarding this problem. I need to ensure that all messages are processed and that there is no duplicity in my system. So I need to take intou account this case (ack timeout), disconnection from rabbitmq, and stopping my application.

In this case, I'm creating a new channel with a new consumer to continue processing messages, and if the message is resended I do not process it, just ack it. I do this because the consumer that generated the timeout is still executing as is processing the prefetched messages.

If in this state I stop my application, How do I know that the first consumer finished processing the prefeteched messages?

[pabermod]

If stop my application I see that connection.Close() is waiting for the first consumer to process all messages

[lukebakken]

@pabermod - I have added your code and an advanced.config for RabbitMQ here

There is much more code than necessary to demonstrate the issue. In addition, there is no code that publishes messages. This means that you are asking me to simplify your code to the bare minimum to demonstrate the issue, as well as figure out how you're publishing messages.

What I'm going to do now is edit the code to the bare minimum to re-create a consumer after a channel timeout, and add a publisher.

[lukebakken]

No, this code always works correctly. I don't know why you're seeing what you saw. Use my code as a starting point.

Notice that I'm using reset events / handles to coordinate threads rather than checking IsOpen, etc.

[pabermod]

Hi, i'm not managing to reproduce the issue again with my code. I executed a load environment that receives messages last weekend and got the timeout there:

System.TimeoutException: The operation has timed out.
at RabbitMQ.Util.BlockingCell`1.WaitForValue(TimeSpan timeout)
at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)
at RabbitMQ.Client.Framing.Impl.Model._Private_ChannelOpen(String outOfBand)
at RabbitMQ.Client.Framing.Impl.Connection.CreateModel()
...

Maybe some kind of race condition or lack of resources?

[pabermod]

Hi, i managed to reproduce the issue with your code:

• set consumer_timeout = 2000
• Use quorum queue with Topic exchange
• In the consumer instead of creating a Timer, use Thread.Sleep(r.Next(1000, 10000));

(The timer is waiting for the elapsed time on another thread so it does not represent a sync operation that takes some time)

[pabermod]

Also i'm seeing some strange behaviour: I added log before/after calling connection.CreateModel() and I see the following:

IModel CreateModel(IConnection connection)
{
    channelShutdownEvent.Reset();
    Console.WriteLine($"[{Environment.CurrentManagedThreadId}][{DateTime.Now:HH:mm:ss.fff}] Creating channel");
    var ch = connection.CreateModel();
    ch.BasicQos(0, 5, false);
    ch.ModelShutdown += OnModelShutdown;
    ch.CallbackException += OnModelCallbackException;
    ch.BasicAcks += OnModelBasicAcks;
    ch.BasicNacks += OnModelBasicNacks;
    ch.BasicReturn += OnModelBasicReturn;
    ch.FlowControl += OnModelFlowControl;
    Console.WriteLine($"[{Environment.CurrentManagedThreadId}][{DateTime.Now:HH:mm:ss.fff}] channel created");
    return ch;
}

The new channel is not created until all prefeteched messages of old consumer are processed... I don't see why this happengs, If i'm opening other indepent channels while i'm receiving messages will they this time to open?

The first channel creation was almost instant...

[lukebakken]

Thanks for the update.

Yes, you're probably getting the following due to excessive load on your cluster:

System.TimeoutException: The operation has timed out.
at RabbitMQ.Util.BlockingCell`1.WaitForValue(TimeSpan timeout)

The general best practice is to NOT do long-running work (aka Sleep) in you consumer callback.

I will add the additional logging to my code with regard to channel creation.

Again, I'm busy with higher-priority work but will return to this when I can.

If you're trying to solve an issue with consumer timeouts, your best option at this time would be to set the timeout in RabbitMQ to a value greater than any you expect in your environment.