How do I recover from UnroutableError after RabbitMQ is restarted?

[alanbchristie]

When I restart RabbitMQ otehr Pods in my cluster (specifically Python Pods) that had connected to the service encounter NO ROUTE errors. No matter what I do I do not seem to be able to recover from the problem without restarting the client (Python) Pod. But, surely, I should be able to recover from the problem without restarting the client Pod?

I've used pika and the official rabbit ampq driver, both behave the same. In the discussion there (pika/pika#1543) the concern was why I don't get a connection failure: -

1. I establish a BlockingConnection connection successfully.
2. I force RabbitMQ to restart (kill the Pod)
3. The connection fails with the expected UnroutableError
4. Any attempt to close and re-establish the connection fails with continual UnroutableError errors

[lukebakken]

Hi @alanbchristie

Both pika/pika#1543 and this issue are missing important information - specifically, the RabbitMQ and Erlang version you're using, as well as details about your RabbitMQ topology and application code

Again, your symptoms point to something specific in your application code or environment. What happens when you run your application against a single-node RabbitMQ running outside of k8s? I suspect you don't see this issue.

Here are some assumptions I've made (please correct them if wrong):

• You are using the latest RabbitMQ docker image in your k8s environment.
• Your pika application is publishing to a quorum queue.
• Your pika application has publisher confirmations enabled (i.e. confirm_delivery has been called). I say this because pika will only raise UnroutableError in this case:

          :raises UnroutableError: raised when a message published in
          publisher-acknowledgments mode (see
          `BlockingChannel.confirm_delivery`) is returned via `Basic.Return`
          followed by `Basic.Ack`.
  

[alanbchristie]

OK, so here’s the sequence of method call in my topic consumer (which is essentially a copy of the reference example): -

1. base_connection._proto_connection_lost() with the ERROR connection_lost: StreamLostError: ("Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')",)
2. on_channel_closed() where I issue the WARNING Channel 1 was closed with the reason Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')
3. close_connection() where I issue the INFO message Connection is closing or already closed because self._connection.is_closing or self._connection.is_closed
4. Finally, on_connection_closed() where I issue the WARNING Connection closed, reconnect necessary, again, with the reason Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')

My on_channel_closed()...

def on_channel_closed(self, channel, reason):
    """Invoked by pika when RabbitMQ unexpectedly closes the channel.
    Channels are usually closed if you attempt to do something that
    violates the protocol, such as re-declare an exchange or queue with
    different parameters. In this case, we'll close the connection
    to shutdown the object.

    :param pika.channel.Channel: The closed channel
    :param Exception reason: why the channel was closed
    """
    _LOGGER.warning('Channel %i was closed: %s', channel, reason)
    self.close_connection()

and close_connection()...

def close_connection(self):
    self._consuming = False
    if self._connection.is_closing or self._connection.is_closed:  # type: ignore
        _LOGGER.info('Connection is closing or already closed')
    else:
        _LOGGER.info('Closing connection')
        self._connection.close()  # type: ignore

and on_connection_closed()...

def on_connection_closed(self, _unused_connection, reason):
    """This method is invoked by pika when the connection to RabbitMQ is
    closed unexpectedly. Since it is unexpected, we will reconnect to
    RabbitMQ if it disconnects.

    :param pika.connection.Connection connection: The closed connection obj
    :param Exception reason: exception representing reason for loss of
        connection.
    """
    self._channel = None
    if self._closing:
        self._connection.ioloop.stop()  # type: ignore
    else:
        _LOGGER.warning('Connection closed, reconnect necessary: %s', reason)
        self.reconnect()

and, finally, reconnect() is...

def reconnect(self):
    """Will be invoked if the connection can't be opened or is
    closed. Indicates that a reconnect is necessary then stops the
    ioloop.
    """
    self.should_reconnect = True
    self.stop()

I guess the damage is done in stop()?

Which is seen issuing the messages Consuming - stopping and Stopped as it stops the ioloop...?

def stop(self):
    """Cleanly shutdown the connection to RabbitMQ by stopping the consumer
    with RabbitMQ. When RabbitMQ confirms the cancellation, on_cancelok
    will be invoked by pika, which will then closing the channel and
    connection.
    """
    _LOGGER.info('Stop requested...')
    if self._closing:
        _LOGGER.info('Already closing - ignoring')
    else:
        self._closing = True
        if self._consuming:
            _LOGGER.info('Consuming - stopping')
            self._connection.ioloop.stop()  # type: ignore
            _LOGGER.info('Stopped')
        else:
            _LOGGER.info('Not consuming - nothing to stop')
    _LOGGER.info('Stop actioned')

    # Kill ourselves (to force a restart of the Pod)
    _LOGGER.info('Poisoning ourselves to force a restart...')
    take_poison()

For the time-being I now rely on 'poison' mechanism that our Pods implement. It is triggered under this condition and causes the Kubernetes Pod's liveness probe to fail. When this happens there is an automatic restart of the Pod. With this poison logic in place the error is at least handled by a restart of the Pod (Container).