3.10.3 apache-airflow-providers-celery release is broken

[zaibacu]

Apache Airflow version

2.10.5

If "Other Airflow 2 version" selected, which one?

No response

What happened?

Version: 3.10.3 release of apache-airflow-providers-celery (released March 13) doesn't work at all with Airflow version 2.10.5 (and 2.10.4). ALL of the tasks instantly fails.

The exception is:execute_command() takes 1 positional argument but 9 were given inside Airflow Scheduler

What you think should happen instead?

Celery Executor shouldn't fail

How to reproduce

Running literally any task while using CeleryExecutor

Operating System

Debian 12 bookworm

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[ms32035]

Same here. this is even more annoying as we have to stick to CeleryKubernetesExecutor, because the helm chart 1.16 which is ready for some time and supports multiple executors still isn't being released

[karenbraganz]

I tested this out and am also seeing the error you mentioned. The task is added to queue but fails before it can start running.

[2025-03-14T22:25:46.102+0000] {scheduler_job_runner.py:692} INFO - Sending TaskInstanceKey(dag_id='sla_breach_warning', task_id='get_opened_ticket_id', run_id='manual__2025-03-14T22:25:45.018635+00:00', try_number=1, map_index=-1) to CeleryExecutor with priority 17 and queue default

[2025-03-14T22:25:46.102+0000] {base_executor.py:169} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'sla_breach_warning', 'get_opened_ticket_id', 'manual__2025-03-14T22:25:45.018635+00:00', '--local', '--subdir', 'DAGS_FOLDER/sla_breach_warning_dag.py']

[2025-03-14T22:25:46.134+0000] {celery_executor.py:312} ERROR - Error sending Celery task: execute_command() takes 1 positional argument but 9 were given
Celery Task ID: TaskInstanceKey(dag_id='sla_breach_warning', task_id='get_opened_ticket_id', run_id='manual__2025-03-14T22:25:45.018635+00:00', try_number=1, map_index=-1)
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/celery/executors/celery_executor_utils.py", line 270, in send_task_to_executor
    result = task_to_run.apply_async(args=args, queue=queue)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/celery/app/task.py", line 559, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: execute_command() takes 1 positional argument but 9 were given

[2025-03-14T22:25:46.134+0000] {scheduler_job_runner.py:776} INFO - Received executor event with state failed for task instance TaskInstanceKey(dag_id='sla_breach_warning', task_id='get_opened_ticket_id', run_id='manual__2025-03-14T22:25:45.018635+00:00', try_number=1, map_index=-1)

[tirkarthi]

In 984c61d I can see the below change where args used to be a single item list of command changed to args that celery tries to check that could be causing this? This was merged to 3.10.2 but it got yanked.

@@ -213,15 +257,19 @@ def send_task_to_executor(
     task_tuple: TaskInstanceInCelery,
 ) -> tuple[TaskInstanceKey, CommandType, AsyncResult | ExceptionWithTraceback]:
     """Send task to executor."""
-    key, command, queue, task_to_run = task_tuple
+    from airflow.executors import workloads
+
+    key, args, queue, task_to_run = task_tuple
+    if isinstance(args, workloads.BaseWorkload):
+        args = (args.model_dump_json(),)
     try:
         with timeout(seconds=OPERATION_TIMEOUT):
-            result = task_to_run.apply_async(args=[command], queue=queue)
+            result = task_to_run.apply_async(args=args, queue=queue)
     except (Exception, AirflowTaskTimeout) as e:
         exception_traceback = f"Celery Task ID: {key}\n{traceback.format_exc()}"
         result = ExceptionWithTraceback(e, exception_traceback)
 
-    return key, command, result
+    return key, args, result

cc: @ashb @eladkal

[ashb]

🤦🏻

[potiuk]

Indeed 🤦 ... I think that underlines the need of completing this one #47518 -> currently our k8s tests only really test K8S executor, not the other executors - they literally just deploy the executor-configured airflow to K8S but no actual tests are run.

[potiuk]

I will do it after moving airflow-core.

[EduardoCastanho]

hi! In what airflow previous version this would be working?
thanks

[EduardoCastanho]

hi! In what airflow previous version this would be working? thanks

downgraded apache-airflow-providers-celery
pip install apache-airflow-providers-celery==3.10.0

[eladkal]

@potiuk was this fixed? did I miss it? I plan to start cut new wave