This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f74a685978c Use configured default queue for Celery executor callbacks
(#73552)
f74a685978c is described below
commit f74a685978c1048e115c94abde445075089b3953
Author: ccoliu <[email protected]>
AuthorDate: Mon Oct 5 08:22:29 2026 +0800
Use configured default queue for Celery executor callbacks (#73552)
CeleryExecutor hardcoded the "default" queue when dispatching
ExecuteCallback workloads, ignoring [operators] default_queue. In
deployments without a Celery queue named "default", deadline
SyncCallbacks were never picked up by a worker.
Read the queue from the executor's (team-aware) config, matching how the
Edge executor routes callbacks. A queue set in the callback data still
takes precedence.
closes: #73545
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../providers/celery/executors/celery_executor.py | 4 +--
.../unit/celery/executors/test_celery_executor.py | 29 ++++++++++++++++++----
2 files changed, 26 insertions(+), 7 deletions(-)
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
index 68304cdbea0..91d7509da25 100644
--- a/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
+++ b/providers/celery/src/airflow/providers/celery/executors/celery_executor.py
@@ -191,8 +191,8 @@ class CeleryExecutor(BaseExecutor):
key = self.get_task_key(workload.ti) if
self.supports_task_instance_uuid else workload.ti.key
workloads_to_be_sent.append((key, workload, workload.ti.queue,
self.team_name))
elif AIRFLOW_V_3_2_PLUS and isinstance(workload, ExecuteCallback):
- # Use default queue for callbacks, or extract from callback
data if available.
- queue = "default"
+ # Use the configured default queue for callbacks, or extract
from callback data if available.
+ queue = self.conf.get_mandatory_value("operators",
"default_queue")
if isinstance(workload.callback.data, dict) and "queue" in
workload.callback.data:
queue = workload.callback.data["queue"]
workloads_to_be_sent.append((workload.callback.key, workload,
queue, self.team_name))
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 4579fa436eb..bbd1493b32f 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -1216,18 +1216,35 @@ def test_celery_tasks_registered_on_import():
@pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS, reason="ExecuteCallback requires
Airflow 3.2+")
@pytest.mark.parametrize(
- ("callback_data", "expected_queue"),
+ ("callback_data", "configured_default_queue", "expected_queue"),
[
- pytest.param({"path": "tests.callbacks.test_callback", "kwargs": {}},
"default", id="default_queue"),
+ pytest.param(
+ {"path": "tests.callbacks.test_callback", "kwargs": {}}, None,
"default", id="default_queue"
+ ),
pytest.param(
{"path": "tests.callbacks.test_callback", "kwargs": {}, "queue":
"callback_queue"},
+ None,
"callback_queue",
id="callback_queue",
),
+ pytest.param(
+ {"path": "tests.callbacks.test_callback", "kwargs": {}},
+ "custom_default_queue",
+ "custom_default_queue",
+ id="configured_default_queue",
+ ),
+ pytest.param(
+ {"path": "tests.callbacks.test_callback", "kwargs": {}, "queue":
"callback_queue"},
+ "custom_default_queue",
+ "callback_queue",
+ id="callback_queue_overrides_configured_default_queue",
+ ),
],
)
@mock.patch("airflow.providers.celery.executors.celery_executor.CeleryExecutor._send_workloads")
-def test_process_workloads_routes_execute_callback(mock_send_workloads,
callback_data, expected_queue):
+def test_process_workloads_routes_execute_callback(
+ mock_send_workloads, callback_data, configured_default_queue,
expected_queue
+):
"""CeleryExecutor routes callback workloads to Celery with the expected
queue."""
from airflow.executors import workloads
from airflow.executors.workloads.callback import CallbackDTO
@@ -1245,8 +1262,10 @@ def
test_process_workloads_routes_execute_callback(mock_send_workloads, callback
log_path="callback.log",
)
- executor = celery_executor.CeleryExecutor()
- executor._process_workloads([workload])
+ config = {("operators", "default_queue"): configured_default_queue} if
configured_default_queue else {}
+ with conf_vars(config):
+ executor = celery_executor.CeleryExecutor()
+ executor._process_workloads([workload])
mock_send_workloads.assert_called_once_with([(workload.callback.key,
workload, expected_queue, None)])
