Consumers do not resume on AutoRecovery because ConsumerWorkService is Shutdown

[jag919]

Looks like in 3.5.0 HandleBasicDeliver is delegated to the dispatcher. Trouble seems to be when the connection was closed, Shutdown was called on the work service, clearing all the keys out. Nothing re-registers the keys with the work service when the connection is restored.

[RabbitMQ.Client.Impl.RecoveryAwareModel] {Session#1:Connection(aa6ca86e-62e9-4a25-aac2-dd2a6fc66aa7,amqp://localhost:5672)} RabbitMQ.Client.Impl.RecoveryAwareModel

The given key was not present in the dictionary.
at System.Collections.Generic.Dictionary2.get_Item(TKey key) at RabbitMQ.Util.BatchingWorkPool2.AddWorkItem(K key, V item) in projects\client\RabbitMQ.Client\src\util\BatchingWorkPool.cs:line 65

[michaelklishin]

By "connection was closed" you mean "connection was lost", that is, not closed deliberately by the app, correct?

[jag919]

Correct, I forced a connection closed from the management console. the connection then recovered, redeclared its models and topology. The message gets delivered to the client, but never handed to the consumer.

[michaelklishin]

Thanks, we'll investigate.

[michaelklishin]

@jag919 I've added a test in 134905d that does not reproduce the problem. We'll try with some manual testing.

[michaelklishin]

Manual testing with a couple of F# scripts also doesn't reveal the problem.

Publisher

#r "RabbitMQ.Client.dll"

open System
open RabbitMQ.Client
open System.Text
open System.Collections.Generic

let enc = Encoding.UTF8
let rnd = new Random()

let cf = new ConnectionFactory()
let conn = cf.CreateConnection()
let ch = conn.CreateModel()
let q = "dotnet-issue-50"

ch.QueueDeclare(q, true, false, false, null) |> ignore

for i in 1 .. 100 do
    let msg = "Message " + i.ToString()
    let body = enc.GetBytes(msg)
    ch.BasicPublish("", q, null, body)
    printfn " [x] Sent %s" msg

conn.Close()

Consumer

#r "RabbitMQ.Client.dll"

open System
open RabbitMQ.Client
open System.Text
open System.Collections.Generic
open System.Threading

let enc = Encoding.UTF8
let rnd = new Random()

let cf = new ConnectionFactory(AutomaticRecoveryEnabled = true, UseBackgroundThreadsForIO = false)
let conn = cf.CreateConnection()
let ch = conn.CreateModel()
let q = "dotnet-issue-50"

ch.QueueDelete(q)
ch.QueueDeclare(q, true, false, false, null) |> ignore

type MyConsumer(ch: IModel) =
    inherit DefaultBasicConsumer(ch)
    override this.HandleBasicDeliver(consumerTag, deliveryTag, redelivered, exchange,
                                     routingKey, props, body) =
        let msg = enc.GetString(body)
        printfn "=> %s" msg


ch.BasicConsume(q, true, new MyConsumer(ch))

let rec loop(): unit =
    Thread.Sleep(5000)
    loop()

loop()

Steps Taken to Reproduce

1. Start consumer.
2. Start publisher.
3. Check that consumer has consumed 100 messages.
4. Stop RabbitMQ.
5. Start RabbitMQ in 5-10 seconds.
6. Check that consumer's reconnected in the management UI.
7. Start publisher.
8. Check that consumer has consumed 100 messages.

[michaelklishin]

Inspecting the code suggests that every AutorecoveringConnection upon reconnection gets a new Connection instance, which in turn sets up a ConsumerWorkService instance, which instantiates BatchingWorkPool. So all should be re-initialised.