Kombu: Does message confirmation work? (Rabbit backend)

[anentropic]

I see there's a commit referenced in #193 and some stuff in the docs.

I have set up an errback function and am using ensure, have also tried the extra transport options given here: http://stackoverflow.com/a/21016469/202168

publish appears to send a message to my broker: I see a blip in the 'Return' graph of my test exchange (Rabbit web UI) but nothing goes through to the queue.

I guess I have something configured wrong, it sounds to me like a 'Return' means the message has bounced. My errback does not get called though.

I also noticed that here in publish method it says the mandatory and immediate args are "currently not supported". I tried using them and immediate gives a "AMQPNotImplementedError" but mandatory does not.

At the moment I'm not sure a) if the kombu message confirm/retry/errback stuff works or b) how to debug my publishing problem.

I was also wondering about the on_return arg for a Producer ("Callback to call for undeliverable messages")... it sounds like what I need. It says "Note that the producer needs to drain events to use this feature"... I'm not clear exactly how this relates to the producer.

There's a drain_events method on the connection... I tried:

publish = conn.ensure(producer, producer.publish,
                      errback=_errback, max_retries=3)

publish(message, **kwargs)
print conn.drain_events(timeout=5)

The drain times-out though... am I doing it wrong? or no return/error means no events to drain, i.e. message succeeded?

[anentropic]

wondering if relevant https://github.com/celery/librabbitmq/issues/25

[bofm]

@anentropic
Here is a dirty workaround for synchronously handling unroutable message returns with pyamqp transport.

import queue

producer.publish(
    message,
    exchange=exchange,
    routing_key=routing_key,
    durable=True,
    delivery_mode=2,  # Persistent
    mandatory=True
)
try:
    # Unroutable messages are returned and put into producer.channel.returned_messages
    returned_msg = producer.channel.returned_messages.get_nowait()
    raise RuntimeError('Message returned: %s' % repr(returned_msg))
    # RuntimeError: Message returned: basic_return_t(reply_code=312, reply_text='NO_ROUTE', exchange='reports', routing_key='new_report', message=<amqp.basic_message.Message object at 0x104fe2eb8>)
except queue.Empty:
    pass

https://www.rabbitmq.com/confirms.html
If the message is also published as mandatory, the basic.return is sent to the client before basic.ack. The same is true for negative acknowledgements (basic.nack).

Here amqp puts returned messages to producer.channel.returned_messages
https://github.com/celery/py-amqp/blob/10ab8857d716d6f62ca57bf25aa286188b6671a4/amqp/channel.py#L2369

[ask]

@bofm I'm not sure this actually works :(

The publish call is asynchronous so the problem with mandatory is that you also need to wait
for a possible returned message.

If your program is using an async I/O event loop then that is not a problem, but if your program
is synchronous then you have to block for some time to wait for a possible "basic.return" frame.
How do you decide how long to wait? Noting that If it times out you cannot know if the message
was delivered or not.

So for your example to work you need to 1) drain events, 2) look for returned messages 3) time out if takes too long.

[ask]

Btw, "confirms" usually refer to publisher confirmations, which is not related to the mandatory flag.

To use publisher confirms you need to use pyamqp (not librabbitmq), and set the confirm_publish transport option: Connection(transport_options={'confirm_publish': True}), or if using Celery: BROKER_TRANSPORT_OPTIONS = {'confirm_publish': True}.

[bofm]

Return is sent (if sent) before ack or nack. If confirm_publish is set, pyamqp blocks to wait for ack or nack. Publish confirm doesn't guarantee that the message is queued. The message might be confirmed and dropped on the broker side, if no route is found. In terms of business logic the only way to make sure a message is put in a queue on the broker side is to set mandatory flag, check for return and get an ack.

UPD1: confirms are not shown in Rabbit Management Console.

[anentropic]

Thanks, this is very helpful info

[zbentley]

UPD1: confirms are not shown in Rabbit Management Console.

They are shown, at least in RabbitMQ 3.6+; if you open a connection object you can see a "mode" setting of C for connections with confirms enabled.