apache-airflow-providers-celery v 3.10.2 issue import workloads

[MathieuB1]

Apache Airflow version

2.10.5

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

2.10.3

What happened?

Hi everyone,

Airflow cannot import correctly all dags, the schduler is stuck with the following error:

[2025-02-27T08:16:39.671+0100] {scheduler_job_runner.py:1017} INFO - Exited execute loop
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 8, in <module>
    sys.exit(main())
             ^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/__main__.py", line 62, in main
    args.func(args)
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/cli_config.py", line 49, in command
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/cli.py", line 116, in wrapper
    return f(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/providers_configuration_loader.py", line 55, in wrapped_function
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 59, in scheduler
    run_command_with_daemon_option(
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/daemon_utils.py", line 86, in run_command_with_daemon_option
    callback()
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 62, in <lambda>
    callback=lambda: _run_scheduler_job(args),
                     ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/cli/commands/scheduler_command.py", line 48, in _run_scheduler_job
    run_job(job=job_runner.job, execute_callable=job_runner._execute)
  File "/usr/local/lib/python3.12/site-packages/airflow/utils/session.py", line 97, in wrapper
    return func(*args, session=session, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/jobs/job.py", line 421, in run_job
    return execute_job(job, execute_callable=execute_callable)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/jobs/job.py", line 450, in execute_job
    ret = execute_callable()
          ^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 987, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1132, in _run_scheduler_loop
    executor.heartbeat()
  File "/usr/local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 58, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 243, in heartbeat
    self.trigger_tasks(open_slots)
  File "/usr/local/lib/python3.12/site-packages/airflow/traces/tracer.py", line 58, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/executors/base_executor.py", line 382, in trigger_tasks
    self._process_tasks(task_tuples)
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/celery/executors/celery_executor.py", line 265, in _process_tasks
    self._send_tasks(task_tuples_to_send)
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/celery/executors/celery_executor.py", line 290, in _send_tasks
    key_and_async_results = self._send_tasks_to_celery(task_tuples_to_send)
                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/celery/executors/celery_executor.py", line 329, in _send_tasks_to_celery
    return list(map(send_task_to_executor, task_tuples_to_send))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/airflow/providers/celery/executors/celery_executor_utils.py", line 260, in send_task_to_executor
    from airflow.executors import workloads
ImportError: cannot import name 'workloads' from 'airflow.executors' (/usr/local/lib/python3.12/site-packages/airflow/executors/__init__.py)

I need to downgrade to apache-airflow-providers-celery v3.10.0.

Can you please investigate?

Many thanks in advance

What you think should happen instead?

No response

How to reproduce

Install airflow 2.10.3 with the following deps:

airflow-exporter==1.5.3
apache-airflow[celery]==2.10.3
apache-airflow[statsd]==2.10.3

Deps for Airflow extracted from https://raw.githubusercontent.com/apache/airflow/constraints-2.8.1/constraints-3.9.txt

Flask-Session==0.5.0

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

apache-airflow-providers-celery==3.10.0

Deployment

Official Apache Airflow Helm Chart

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.

[eladkal]

cc @ashb I believe this is a regression due to #46265 ?

[eladkal]

I yanked celery provider 3.10.2 and we are working on a fix for this bug