How can I perform a long task

[dabbler0606]

We just recently upgraded from RabbitMQ 3.3.5 to 3.8.27

It is found that some long-running tasks (about 1 hour) will trigger message requeue

So I adjusted consumer_timeout to 24 hours

But the message will still trigger redelivery in about 3 minutes. Why is this

I tried to adjust the heartbeat to 0, 3600 and found that it still didn't work

Thank you every one !

Producer

import pika


class Producer:
    def __init__(self):
        params = pika.ConnectionParameters()
        connection = pika.BlockingConnection(params)
        self._channel = connection.channel()

    def publish(self, body):
        self._channel.queue_declare('queue1')
        self._channel.exchange_declare('queue1')
        self._channel.queue_bind('queue1', 'queue1')
        self._channel.basic_publish('queue1', 'queue1', body)


if __name__ == '__main__':
    for i in range(100):
        print(i)
        Producer().publish('hello world')

Consumer

import pika
import time


class Consumer:
    def __init__(self):
        params = pika.ConnectionParameters()
        params = pika.ConnectionParameters()
        connection = pika.BlockingConnection(params)
        self._channel = connection.channel()

    def start(self):
        def on_message_callback(channel, method, properties, body):
            print(body)
            for i in range(3600):
                print(i)
                time.sleep(1)

            print('回复 ack')
            self._channel.basic_ack(method.delivery_tag)

        self._channel.basic_qos(prefetch_count=50)
        self._channel.basic_consume('queue1', on_message_callback)
        self._channel.start_consuming()


if __name__ == '__main__':
    Consumer().start()

[michaelklishin]

We cannot suggest much without any server logs. A traffic capture would tell you exactly what both sides send.

If you perform a blocking operation in the consumer handler, with some clients this may block all outgoing I/O and result in missed heartbeats detected by RabbitMQ. The default heartbeat timeout is 60 seconds and not three minutes, though.

This is a Pika-specific question.

[michaelklishin]

Also, if a consumer delivery hits a timeout, there will be a specific log message. The default is 30 minutes, not 3.

You can inspect your node's effective configuration as it may actually be different from what you think it is.

[michaelklishin]

Finally, RabbitMQ 3.8 is about to go completely unsupported.

While 3.8.27 is an improvement over 3.3.5 from August 2014, you are five patch releases behind in a release series that's a few weeks away from EOL.

Consider upgrading to 3.10.

[Gsantomaggio]

It is documented here:
https://pika.readthedocs.io/en/stable/examples/heartbeat_and_blocked_timeouts.html

Here the example:
https://github.com/pika/pika/blob/main/examples/heartbeat_and_blocked_timeouts.py

import pika
import time
from datetime import datetime


class Consumer:
    def __init__(self):
        params = pika.ConnectionParameters(heartbeat=600, blocked_connection_timeout=300)
        connection = pika.BlockingConnection(params)
        self._channel = connection.channel()

    def start(self):
        start = datetime.now() 
        def on_message_callback(channel, method, properties, body):
            print(body)
            for i in range(3600):
                print("seconds: {} id: {}".format((datetime.now()-start).total_seconds(), i) )
                time.sleep(1)

            print('回复 ack')
            self._channel.basic_ack(method.delivery_tag)

        self._channel.basic_qos(prefetch_count=50)
        self._channel.basic_consume('queue1', on_message_callback)
        self._channel.start_consuming()


if __name__ == '__main__':
    Consumer().start()

I left it up and running for a while:

seconds: 1356.463489 id: 1352
seconds: 1357.466798 id: 1353
seconds: 1358.470838 id: 1354
seconds: 1359.472617 id: 1355

The connection was still up and running:

➜  rabbitmqctl list_connections
Listing connections ...
user    peer_host       peer_port       state
guest   ::1     58917   running

[dabbler0606]

If this is done, an error is reported at the end

2022-07-05 14:37:00.421 [info] <0.5963.2> accepting AMQP connection <0.5963.2> (172.17.0.1:55062 -> 172.17.0.2:5672)
2022-07-05 14:37:00.433 [info] <0.5963.2> connection <0.5963.2> (172.17.0.1:55062 -> 172.17.0.2:5672): user 'guest' authenticated and granted access to vhost '/'
2022-07-05 15:07:00.446 [error] <0.5963.2> closing AMQP connection <0.5963.2> (172.17.0.1:55062 -> 172.17.0.2:5672):
missed heartbeats from client, timeout: 600s
2022-07-05 15:07:00.449 [info] <0.6644.2> Closing all channels from connection '172.17.0.1:55062 -> 172.17.0.2:5672' because it has been closed

[dabbler0606]

I set the task execution time to 1800s

I'm using a local docker, I don't know if there's an impact

[Gsantomaggio]

Per conversation with @dabbler0606: setting HeartBeat to 0 solves the issue.
Another way could be: https://github.com/pika/pika#requesting-message-acknowledgements-from-another-thread

[michaelklishin]

Disabling heartbeats is just wrong. Concurrent execution of such long running operations is the right way to go about it.

[Gsantomaggio]

Disabling heartbeats is just wrong

100%. we already spoke about that.