Rabbitmq consumer crashes once rabbitmq server recovered from failure

[raj-odc]

Hello Guys,
I am facing a problem with crashing the rabbitMQ consumer during the restart of the AMQP server.
My rails application with following versions:
ruby -2.6.6
Rails - 5.2
bunny-2.22.0 -> with automatic_retry: true and no retry limit set

The issue in details:

1. I start my rabbitmq consumer and start consuming the messages from the queue
2. I kill the rabbitMQ server, consumer keeps on retrying with Retrying connection on next host in line: 127.0.0.1:5672, during this period consumer will not close
3. I started the rabbitMQ server again, and now the consumer goes down. Automatically kills the consumer.

[michaelklishin]

I'm afraid that's not at all "in detail". A detailed report would be an executable way to reproduce, or at least an example app with a set of step-by-step instructions about how exactly the connection is set up, and so on.

You cannot claim that a "crashing consumer" (whatever exactly that means) is a responsibility of Bunny. You are responsible for how your own applications react to node failures. Bunny offers a part of a solution that is covered in the error handling guide.

In my tests, Bunny recovers just fine, after repeated recoveries and RabbitMQ node shutdown. In fact, I observe this regularly during RabbitMQ's own development where I sometimes use Bunny connections from a REPL, and they recover without failure every time I recompile and restart the local node.

Having an exception in the log does not mean that the "consumer crashes". There will be an exception and errors logged when a connection fails to reconnect, by design.

Finally, there are recovery-related settings that limit the number of recovery attempts
and control whether the limit is "global" or should be reset upon each recovery.

By default there is no limit and the recovery times counter is reset.

[raj-odc]

@michaelklishin Thanks for your response,

I have set up all the recovery settings properly and it works perfectly for the Puma service.

Failure happens only on the daemon service which I have for consuming the rabbitMq message and processing it.

I will try to reiterate the problem,

1. I begin consuming messages from the queue using my RabbitMQ consumer code given below.
2. If I terminate the RabbitMQ server intentionally to replicate this issue, the consumer continues attempting to reconnect, displaying a message like "Retrying connection on next host in line: 127.0.0.1:5672." During this phase, the consumer remains active without closing.

Until this point, it's working perfectly as expected

3. Upon starting again the RabbitMQ server, the consumer stops instead of reconnecting and processing the messages.

Below is the code for consuming the messages from the rabbitmq queue.

# encoding: UTF-8
# frozen_string_literal: true

require File.join(ENV.fetch('RAILS_ROOT'), 'config', 'environment')

raise "bindings must be provided." if ARGV.size == 0

logger = Rails.logger

conn = Bunny.new AMQP::Config.connect
conn.start

ch = conn.create_channel
id = $0.split(':')[2]
prefetch = AMQP::Config.channel(id)[:prefetch] || 0
ch.prefetch(prefetch) if prefetch > 0
logger.info { "Connected to AMQP broker (prefetch: #{prefetch > 0 ? prefetch : 'default'})" }

terminate = proc do
  # logger is forbidden in signal handling, just use puts here
  puts "Terminating threads .."
  ch.work_pool.kill
  puts "Stopped."
end

at_exit { conn.close }

Signal.trap("INT",  &terminate)
Signal.trap("TERM", &terminate)

workers = []
ARGV.each do |id|
  queue  = ch.queue *AMQP::Config.binding_queue(id)
  clean_start = AMQP::Config.data[:binding][id][:clean_start]

  queue.purge if clean_start
  worker = AMQP::Config.binding_worker(id)

  if args = AMQP::Config.binding_exchange(id)
    x = ch.send *args

    case args.first
    when 'direct'
      queue.bind x, routing_key: AMQP::Config.routing_key(id)
    when 'topic'
      AMQP::Config.topics(id).each do |topic|
        queue.bind x, routing_key: topic
      end
    else
      queue.bind x
    end
  end

  # Enable manual acknowledge mode by setting manual_ack: true.
  queue.subscribe manual_ack: true do |delivery_info, metadata, payload|
    tagged_logger = ::TaggedLogger.new(logger, job_id: SecureRandom.uuid)
    Rails.logger = tagged_logger
    tagged_logger.info { "Received: #{payload}" }
    begin
      # Invoke Worker#process with floating number of arguments.
      args          = [JSON.parse(payload), metadata, delivery_info]
      arity         = worker.method(:process).arity
      resized_args  = arity < 0 ? args : args[0...arity]
      worker.process(*resized_args)

      ch.ack(delivery_info.delivery_tag)
    rescue StandardError => e
       ch.nack(delivery_info.delivery_tag, false, true)

      if worker.is_db_connection_error?(e)
        tagged_logger.error(db: :unhealthy, message: e.message)
        exit(1)
      end

      report_exception(e)
    end
    Rails.logger = logger
  end

  workers << worker
end

%w(USR1 USR2).each do |signal|
  Signal.trap(signal) do
    puts "#{signal} received."
    handler = "on_#{signal.downcase}"
    workers.each {|w| w.send handler if w.respond_to?(handler) }
  end
end

ch.work_pool.join

[michaelklishin]

This involves Puma (a server that uses threads), signal handling, explicit joins with the channels' consumer dispatch work pool (which is not really a part of the public API).

I'm afraid I don't have the cycles to investigate something like this coming from a non-paying user.

[michaelklishin]

Switching Bunny log level to debug should reveal all outgoing and inbound frames.