Distributed python: Celery send_task gets COMMAND_INVALID

1. Context

I developed a Flask API that sends tasks to my computing environment. To use this, you should make a post request to the API. Then, the API received your request, process it and send necessary data, through the RABBITMQ broker, a message to be held by the computing environment. At the end, it should send the result back to the API

2. Some code

Here is an example of my API and my Celery application:

#main.py

# Package
import time
from flask import Flask
from flask import request, jsonify, make_response

# Own module
from celery_app import celery_app

# Environment
app = Flask()

# Endpoint
@app.route("/test", methods=["POST"])
def test():
    """
    Test route

    Returns
    -------
    Json formatted output
    """

    # Do some preprocessing in here 

    result = celery_app.send_task(f"tasks.Client", args=[1, 2])
    while result.state == "PENDING":
        time.sleep(0.01)
    result = result.get()

    if result["sucess"]:
        result_code = 200
    else:
        result_code = 500
    output = str(result)
    return make_response(
        jsonify(
            text=output,
            code_status=result_code,        ),
        result_code,
    )


# Main thread
if __name__ == "__main__":
    app.run()

In a different file, I have setup my celery application connected to RABBITMQ Queue

#celery_app.py
from celery import Celery, Task

celery_app = Celery("my_celery",
                    broker=f"amqp://{USER}:{PASSWORD}@{HOSTNAME}:{PORT}/{COLLECTION}",
                    backend="rpc://"
)
celery_app.conf.task_serializer = "pickle"
celery_app.conf.result_serializer = "pickle"
celery_app.conf.accept_content = ["pickle"]
celery_app.conf.broker_connection_max_retries = 5
celery_app.conf.broker_pool_limit = 1

class MyTask(Task):
    def run(self, a, b):
        return a + b

celery_app.register_task(MyTask())

To run it, you should launch:

python3 main.py

Do not forget to run the celery worker (after registering tasks in it)

Then you can make a post request on it:

curl -X POST http://localhost:8000/test

3. The problem to resolve

When this simple API is running, I am sending request on my endpoint. Unfortunatly, it fails 1 time on 4.

I have 2 messages:

• The first message is:

amqp.exceptions.PreconditionFailed: (0, 0): (406) PRECONDITION_FAILED - delivery acknowledgement on channel 1 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more

• Then, because of the time out, my server has lost the message so:

File "main.py", line x, in test
    result = celery_app.send_task("tasks.Client", args=[1, 2])
amqp.exceptions.InvalidCommand: Channel.close_ok: (503) COMMAND_INVALID - unimplemented method

4. Resolve this error

There are 2 solutions to get around this problem

• retry to send a tasks until it fails 5 times in a row (try / except amqp.exceptions.InvalidCommand)

• change the timeout value.

Unfortunatly, it doesn't seems to be the best ways to solve it.

Can you help me ?

Regards

PS:

• my_packages:

Flask==2.0.2

python==3.6

celery==4.4.5

rabbitmq==latest