Is there a function that allows the celery worker provided by airflow to take DAG from the Rabbitmq Quorum queue?

[jojaeng2]

Apache Airflow version - 2.7.2

RabbitMQ version - 3.12.8

Hello, I'm setting up Airflow using CeleryExecutor with RabbitMQ as the message broker.

After completing the RabbitMQ cluster configuration, I'm working on enabling RabbitMQ High Availability (HA) using quorum queues instead of mirroring, which has been deprecated since version 4.

For testing purposes, I've created a quorum queue in the RabbitMQ Management interface and published a DAG to that queue. However, despite setting up the environment for Celery to consume from the queue, I'm encountering an error where Celery cannot retrieve the DAG from the quorum queue.

PreconditionFailed(406, "PRECONDITION_FAILED - inequivalent arg 'x-queue-type' for queue 'celery' in vhost '/': received none but current is the value 'quorum' of type 'longstr'", (50, 10), 'Queue.declare')

Is the Celery process created as an Airflow celery worker officially unsupported for consuming from quorum queues? Or might there be something I'm overlooking? I'd appreciate any help or guidance on this matter.

[potiuk]

I thikn you might look at https://airflow.apache.org/docs/apache-airflow-providers-celery/stable/configurations-ref.html#celery-config-options to see if you can configure it - I doubt you can get arbitrary answer can/cannot be done - likely you should dive through the celery documentation and see how you can configure celery to make it works. Airlfow merely starts celery and passes the configuration to it, with some things like backend connection configurable directly via airflow configuration and some just passed to celery. also Celry (AFAIK) can be configured additionally via env variables so if the configuraiton options are not enough you might try to do it this way.

Later when (hopefully) you will find how to do it I would encourage you to contribute documentation back on how to do it. When you get to it I am happy to guide you. Airlfow is created by 2700 contributors and such documentation is gladly seen as contribution

[trikker]

Thank you for your kind answer. After looking at various issues and official documents, my conclusion is that Celery currently provided by Airflow does not seem to be able to conceive the Quorum Queue of RabbitMQ or the Stream type Queue. I think an environment variable or an additional config field should be added, but if I have a chance, I will contribute. Thank you again.

Hi, I find a configuration "task_default_queue_type" in celery but it is added in Celery 5.5, its default value is "classic"

https://docs.celeryq.dev/en/latest/userguide/configuration.html#task-default-queue-type

it can be set to "quorum", but the version of celery in airflow(2.10.3, currently newest) is 5.4, so we cannot use it for now. So we need to wait for the version upgrade of Celery in airflow or request the version upgrade of Celery by writing a issue.

[jojaeng2]

@trikker
Oh, thank you for the information. I've been stuck using RabbitMQ 3.xx because quorum queue support wasn't available, but if Airflow upgrades the Celery version soon, I should be able to use RabbitMQ 4.xx. That means I won't have to worry about setting up RabbitMQ clusters or mirroring anymore. Thanks for letting me know!

[trikker]

@trikker Oh, thank you for the information. I've been stuck using RabbitMQ 3.xx because quorum queue support wasn't available, but if Airflow upgrades the Celery version soon, I should be able to use RabbitMQ 4.xx. That means I won't have to worry about setting up RabbitMQ clusters or mirroring anymore. Thanks for letting me know!

Welcome. I am new to RabbitMQ, but I find that RabbitMQ 3.13 supports Quorum Queues, link: https://www.rabbitmq.com/docs/3.13/quorum-queues. I can create quorum queues in the web console of RabbitMQ 3.13, so I think you don't have to upgrade to RabbitMQ 4.xx to use the Quorum Queues. But we need to wait for Airflow to upgrade its Celery version.

[jojaeng2]

Welcome. I am new to RabbitMQ, but I find that RabbitMQ 3.13 supports Quorum Queues, link: https://www.rabbitmq.com/docs/3.13/quorum-queues. I can create quorum queues in the web console of RabbitMQ 3.13, so I think you don't have to upgrade to RabbitMQ 4.xx to use the Quorum Queues. But we need to wait for Airflow to upgrade its Celery version.

Ah, thanks. You're right—I had forgotten since I've been thinking about this issue for a long time. As you mentioned, it's true that I don’t necessarily need to upgrade RabbitMQ to use quorum queues, but there are several reasons why I want to move to version 4.xx. Currently, I’m using version 3.12.xx, which is nearing its EOL. In addition, with version 3.xx, there are too many considerations when setting up clustering, mirroring, and handling network partitions. I believe quorum queues offer a good solution to these issues, which is why I’m eager to upgrade to 4.xx. However, starting from 4.xx, classic queues are no longer supported, so I have to keep 3.xx for compatibility with Airflow.

[trikker]

Welcome. I am new to RabbitMQ, but I find that RabbitMQ 3.13 supports Quorum Queues, link: https://www.rabbitmq.com/docs/3.13/quorum-queues. I can create quorum queues in the web console of RabbitMQ 3.13, so I think you don't have to upgrade to RabbitMQ 4.xx to use the Quorum Queues. But we need to wait for Airflow to upgrade its Celery version.

Ah, thanks. You're right—I had forgotten since I've been thinking about this issue for a long time. As you mentioned, it's true that I don’t necessarily need to upgrade RabbitMQ to use quorum queues, but there are several reasons why I want to move to version 4.xx. Currently, I’m using version 3.12.xx, which is nearing its EOL. In addition, with version 3.xx, there are too many considerations when setting up clustering, mirroring, and handling network partitions. I believe quorum queues offer a good solution to these issues, which is why I’m eager to upgrade to 4.xx. However, starting from 4.xx, classic queues are no longer supported, so I have to keep 3.xx for compatibility with Airflow.

ok, I know your concern of the problem of classic queues. I care about the HA and correctness of MQ too so I tried to use quorum queues too as soon as I deployed RabbitMQ 3.13. Looking forward to the upgrade of Celery in airflow.

[trikker]

@jojaeng2 hi jojaeng2, I am testing the HA of all airflow components and may need your help. After I stop the first node(11.165.218.219) of the rabbitmq cluster with the command rabbitmqctl stop_app, I find the tasks cannot be scheduled. I don't know why airflow cannot use the second rabbitmq node after I stop the first node. I don't know if the below config of broker_url in ariflow can achieve the goal of HA of rabbitmq if any node fails.

RabbitMQ version: 3.13.2
broker url in airflow.cfg: broker_url = amqp://test_user:[email protected]:5672/test_vhost;amqp://test_user:[email protected]:5672/test_vhost;amqp://test_user:[email protected]:5672/test_vhost

queues created by airflow seems don't work as expected:

airflow dag instances fail:

Thank you!

[jojaeng2]

Hi, @trikker
I think it appears to be an issue with the default queue. If you look at the screenshot you shared, you can see that the default queue is down. Airflow DAGs work in a way where tasks are delivered to the default queue created by Celery, and then Celery workers pick up these tasks.

It seems that the default queue in your RabbitMQ cluster is led by the first node, so when that node goes down, tasks cannot be delivered to the default queue, causing scheduling issues. To resolve this, you can use RabbitMQ mirroring settings. You’ll need to set a policy for the default queue, and you can achieve the desired behavior by using the following command for mirroring
rabbitmqctl set_policy ha-all "^default" '{"ha-mode":"all"}'
This should help you achieve the high availability you’re looking for.

[trikker]

Yes, @jojaeng2 you are right, after running this command rabbitmqctl set_policy ha-all "^default" '{"ha-mode":"all"}' -p test_vhost and restarting the rabbitmq node and restarting my celery workers(they crashed because the default queue cannot be connected), my problem is solved, thanks!