Celery executor: per-publish Celery app reconstruction (since providers-celery 3.16.0) causes operation_timeout breaches

[seanmuth]

Apache Airflow Provider(s)

• apache-airflow-providers-celery (regression introduced in 3.16.0, present through current)

Versions of Apache Airflow Providers

• apache-airflow-providers-celery==3.16.0 (introducing release) through 3.19.0 (current) and main. Verified the per-publish create_celery_app(_conf) call in send_workload_to_executor is unchanged across all releases since 3.16.0.
• One partial optimization has landed since: get_celery_configuration() is now @cache-decorated, so the config dict is built once. The Celery() app instance itself is still constructed fresh on every publish, so _backend is still None at the start of every apply_async call and the entry_points() scan still runs per publish. The mechanics of the regression are identical to 3.16.0.
• apache-airflow-providers-cncf-kubernetes is referenced as a counterexample.

Apache Airflow version

Reproduced on Airflow 2.11.0 (Astro Runtime 13.4.0). The relevant code path is present on main as well.

What happened?

Since providers-celery 3.16.0 (PR #60675 — "AIP-67 - Multi Team: Update Celery Executor to support multi team"), send_task_to_executor constructs a fresh Celery() app on every task publish:

# providers/celery/.../celery_executor_utils.py
def send_task_to_executor(task_tuple):
    key, args, queue, team_name = task_tuple
    ...
    celery_app = create_celery_app(_conf)   # NEW: fresh Celery() per publish
    task_to_run = celery_app.tasks["execute_workload"]
    ...
    with timeout(seconds=OPERATION_TIMEOUT):
        result = task_to_run.apply_async(args=args, queue=queue)

The PR's own commit message acknowledges this explicitly:

Since sending tasks is parallelized with multiple processes (which do not share memory with the parent) the send task logic now re-creates a celery app in the sub processes (since the pickling and unpickling that python does to try pass state to the sub processes was not reliably creating the correct celery app objects).

Each Celery() instance starts with _backend = None. The first thing apply_async does is app.send_task(), which accesses self.backend, which lazy-resolves via _get_backend() → backends.by_url() → backends.by_name(), which in turn calls load_extension_class_names("celery.result_backends"), which runs importlib.metadata.entry_points() — walking every installed distribution to read its entry_points.txt and merge plugin-registered backends into the alias map. The scan is run unconditionally; setting [celery] result_backend explicitly does not skip it.

Pre-3.16.0, app = _get_celery_app() was a module-level singleton. The lazy backend resolution happened once per subprocess lifetime, then was cached on the Celery instance. Post-3.16.0, the cache is thrown away every publish, so the scan runs once per task enqueued.

This scan's cost scales linearly with the count of installed distributions on disk. On realistic production images (~600+ *.dist-info entries), each publish now pays a multi-tens-of-milliseconds tax on the quiet path. Under scheduler load and memory pressure, the tail extends past the default 1.0s operation_timeout, causing publishes to fail with AirflowTaskTimeout, exhaust the configured retry budget (default 3), and end up logged as:

{celery_executor.py:174} INFO - [Try 1 of 3] Task Timeout Error for Task: ...
{celery_executor.py:174} INFO - [Try 2 of 3] Task Timeout Error for Task: ...
{celery_executor.py:174} INFO - [Try 3 of 3] Task Timeout Error for Task: ...
{celery_executor.py:213} ERROR - Error sending Celery task: Timeout, PID: NNN

Retries hit the same slow path inside the same subprocess (the new Celery() instance's _backend is still None), so retrying does not recover.

Quantitative impact

Measured on a controlled astro-runtime 13.4.0 deployment matching a real production deployment's scheduler resources (1 vCPU, 2 GiB) and roughly its *.dist-info count (656 distributions). Each iteration mirrors the per-publish path: create_celery_app(conf) + _ = app.backend. Wall-clock latency in milliseconds, n=100 per row:

Configuration	min	p50	p95	p99	max	iter/s
Idle, no load	48.1	51.1	67.0	82.0	86.4	18.8
Idle + 1.5 GiB anon allocation	48.2	52.2	88.8	94.9	100.4	17.5
5 modest DAGs with mapped tasks running	52.2	88.6	232.0	362.7	385.3	9.6
Same DAG load + 1.0 GiB anon allocation	51.0	111.8	498.3	697.9	717.8	5.7

For comparison: a real production deployment with this regression (~682 distributions, ~640 active DAGs, periodic OOMs) shows max=558ms and consistent Task Timeout Error failures at the default 1.0s operation_timeout. The controlled reproduction exceeds that worst case using just 5 DAGs, demonstrating that the regression's impact is multiplicative across three independent factors:

• Installed distribution count (baseline scan cost)
• Scheduler activity / CPU contention with the publisher subprocess
• Memory pressure causing entry_points.txt page-cache eviction

In the field, all three compound simultaneously. The 1.0s operation_timeout default — appropriate for the pre-regression path where backend resolution was amortized — is no longer safe at typical production deployment sizes.

Counterexample: KubernetesExecutor's AIP-67 implementation does not have this problem

PR #61798 implemented the same AIP-67 multi-team feature for KubernetesExecutor. The pattern is cleaner:

# providers/cncf/kubernetes/.../executors/kubernetes_executor.py
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    ...
    self.kube_config = KubeConfig(executor_conf=self.conf)   # once per executor
    ...

KubeConfig is parameterized on the team-aware executor_conf but constructed once, at executor init, and held on the instance. No per-publish reconstruction.

The KubernetesExecutor publishes via a long-lived multiprocessing worker reading from a managed queue. The worker process holds its Kubernetes client across publishes. By contrast, the CeleryExecutor's ProcessPoolExecutor.submit(send_task_to_executor, ...) per-publish design forced the AIP-67 author to push the team-aware app construction inside the per-call function rather than ahead of it — and the pickling concern cited in the commit message was specifically about passing app state through the pool boundary, not about needing fresh apps per task semantically.

In other words, the per-publish reconstruction is not an inherent requirement of multi-team support. It's an artifact of the way the celery publish pool was already shaped. A subprocess-local cache keyed on team_name would satisfy the multi-team correctness requirement without re-paying the backend-resolution cost on every publish.

Proposed fix

Cache the constructed Celery app at module level inside the publisher subprocess, keyed on team:

# providers/celery/.../celery_executor_utils.py
from functools import lru_cache

@lru_cache(maxsize=8)   # one app per active team in a given subprocess
def _cached_celery_app_for_team(team_name: str | None) -> Celery:
    """
    Subprocess-local Celery app cache keyed on team_name.

    Subprocesses don't share memory with the parent, so this cache is per-subprocess.
    Within a subprocess, calls for the same team reuse the same Celery() instance,
    preserving the cached backend resolution that pre-3.16.0 enjoyed module-globally.
    """
    if AIRFLOW_V_3_0_PLUS:
        from airflow.executors.base_executor import ExecutorConf
        _conf = ExecutorConf(team_name)
    else:
        _conf = conf
    return create_celery_app(_conf)


def send_task_to_executor(task_tuple):
    key, args, queue, team_name = task_tuple
    celery_app = _cached_celery_app_for_team(team_name)   # cached, not reconstructed
    if AIRFLOW_V_3_0_PLUS:
        task_to_run = celery_app.tasks["execute_workload"]
        ...
    else:
        task_to_run = celery_app.tasks["execute_command"]
        ...
    with timeout(seconds=OPERATION_TIMEOUT):
        result = task_to_run.apply_async(args=args, queue=queue)

This preserves AIP-67's multi-team correctness (apps are still team-specific where needed) while restoring per-subprocess amortization for the _get_backend() → entry_points() cost. The fix is local to one function, requires no API changes, and aligns the CeleryExecutor's pattern with how KubernetesExecutor handled the same AIP-67 requirement.

The maxsize=8 is a guess at a reasonable per-subprocess team-count ceiling and should be configurable, defaulting to something modest. For single-team deployments (the vast majority), team_name is None and a single cache entry covers all publishes.

How to reproduce

A complete reproduction setup follows. (Happy to share the project skeleton if useful — leaving inline so the steps are self-contained.) Summary:

1. Astro Runtime 13.4.0 (Airflow 2.11.0), 1 vCPU / 2 GiB scheduler.
2. providers-celery==3.17.2, celery==5.6.2.
3. Inflate the image's distribution count to ~656 via requirements.txt (a wide set of unused providers and utility packages — count is what matters, not which ones).
4. Five synthetic DAGs with */1 * * * * and */2 * * * * schedules and mapped tasks producing ~380 task enqueues/minute.
5. Inside the scheduler pod, run a 100-iteration timing script measuring create_celery_app(conf); _ = app.backend. Then layer in a bytearray(1024 * 1024 * 1024) allocation in a second shell. Observe max cross 700ms; observe Task Timeout Error for Task log entries firing.

Downgrade providers-celery to 3.15.x in the same image; the same measurement collapses to roughly the idle baseline regardless of load + pressure, because the entry-points scan is amortized over subprocess lifetime.

Operating system

Reproduction on linux/amd64 (astro-runtime base, Debian-based). Not OS-specific; the regression is in Python-level code.

Deployment

Other 3rd-party Helm chart / managed deployment (Astronomer Hosted Execution). Same root cause is present in any deployment of Airflow 2.11.0+ or 3.x with apache-airflow-providers-celery >= 3.16.0.

Anything else?

• Workarounds that do not help: setting [celery] result_backend explicitly (the entry-points scan runs unconditionally inside by_name); pinning to celery==5.5.x (same code path on Python 3.11+); removing the importlib_metadata backport (it's a transitive apache-airflow dependency).
• Workarounds that do help, in priority: bump [celery] operation_timeout to a generous value (e.g. 2-3s) — this just stops the cliff failure, the per-publish cost remains; downgrade to providers-celery==3.15.x if multi-team is not required.
• Disabling AIRFLOW__CORE__MULTI_TEAM (or running on Airflow 2.x where it doesn't exist) does not mitigate, because the per-publish create_celery_app is unconditional and not gated on whether multi-team is active.
• AIRFLOW__CORE__LAZY_LOAD_PLUGINS has no effect — the scan walks installed distributions on disk, not loaded modules.
• The OOM symptom (memory pressure feedback loop) often shows up alongside this regression in deployments with tight scheduler memory limits; addressing the per-publish cost reduces orphan-adoption-burst load on restart and indirectly helps the OOM picture.

Are you willing to submit a PR?

• Yes

Code of Conduct

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

---

Drafted-by: Claude Code (Opus 4.7); reviewed by @seanmuth before posting

[seanmuth]

@kaxil I'd also like to get the regression agent to identify these. @jedcunningham was asking how this regression made it to release, and I believe any tests that would've touched this codepath likely only hit the "quiet path" and the test wasn't performed with a significant enough number of pip distributions to correctly show the regression