You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "anmolxlight (via GitHub)" <gi...@apache.org> on 2026/05/18 18:27:12 UTC
[PR] Cache Celery apps when publishing workloads [airflow]
anmolxlight opened a new pull request, #67127:
URL: https://github.com/apache/airflow/pull/67127
## What\n\n- cache the Celery app used by inside each publisher subprocess\n- key the cache by so multi-team configurations still get isolated app instances\n- add regression coverage proving repeated publishes for the same team reuse the app while different teams remain separated\n\n## Why\n\n currently creates a fresh app for every publish. Each fresh app loses Celery's lazy backend cache, so repeatedly performs backend resolution and entry point scanning. On large deployments this can push task publishing past .\n\nCaching the app per subprocess preserves the post-AIP-67 behavior of constructing apps inside publisher subprocesses while restoring per-process amortization.\n\nfixes #67123\n\n## Tests\n\n- ============================= test session starts ==============================
platform linux -- Python 3.11.15, pytest-9.0.3, pluggy-1.6.0 -- /home/ubuntu/work/airflow/.venv/bin/python
cachedir: .pytest_cache
rootdir: /home/ubuntu/work/airflow
configfile: pyproject.toml
plugins: cov-7.1.0, anyio-4.13.0, unordered-0.7.0, custom-exit-code-0.3.0, rerunfailures-16.1, xdist-3.8.0, instafail-0.5.0, time-machine-3.2.0, timeouts-1.2.1, requests-mock-1.12.1, asyncio-1.3.0, mock-3.15.1, kgb-7.3, icdiff-0.9
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=function, asyncio_default_test_loop_scope=function
setup timeout: 0.0s, execution timeout: 0.0s, teardown timeout: 0.0s
collecting ... collected 54 items
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_sentry_integration PASSED [ 1%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_supports_sentry SKIPPED [ 3%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_cli_commands_vended PASSED [ 5%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_celery_executor_init_with_args_kwargs PASSED [ 7%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_exception_propagation SKIPPED [ 9%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_gauge_executor_metrics PASSED [ 11%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_command_validation[wrong-command] SKIPPED [ 12%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_command_validation[incomplete-command] SKIPPED [ 14%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_command_validation[complete-command] SKIPPED [ 16%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_task_instances_none SKIPPED [ 18%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_task_instances SKIPPED [ 20%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_with_and_without_external_executor_id SKIPPED [ 22%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_cleanup_stuck_queued_tasks SKIPPED [ 24%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_revoke_task SKIPPED [ 25%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_result_backend_sqlalchemy_engine_options PASSED [ 27%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_operation_timeout_config PASSED [ 29%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_send_workloads_to_celery_hang SKIPPED [ 31%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_send_workload_uses_external_executor_id_as_celery_task_id PASSED [ 33%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_send_workload_reuses_celery_app_for_same_team PASSED [ 35%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_send_workload_keeps_celery_app_cache_team_scoped PASSED [ 37%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_celery_executor_with_no_recommended_result_backend PASSED [ 38%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_sentinel_kwargs_loaded_from_string PASSED [ 40%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_celery_task_acks_late_loaded_from_string PASSED [ 42%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_visibility_timeout_default_warns_when_not_configured PASSED [ 44%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_visibility_timeout_no_warning_when_configured PASSED [ 46%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_visibility_timeout_not_set_for_unsupported_broker PASSED [ 48%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_celery_extra_celery_config_loaded_from_string PASSED [ 50%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_sentinel_kwargs_loaded_from_string PASSED [ 51%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_master_name_loaded PASSED [ 53%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_transport_options_with_multiple_options PASSED [ 55%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_sentinel_kwargs_invalid_json PASSED [ 57%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_sentinel_kwargs_not_dict PASSED [ 59%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_result_backend_sentinel_full_config PASSED [ 61%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestMultiTeamCeleryExecutor::test_multi_team_isolation_and_workload_routing PASSED [ 62%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestMultiTeamCeleryExecutor::test_workload_routing_through_team_specific_app PASSED [ 64%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_celery_tasks_registered_on_import PASSED [ 66%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_process_workloads_routes_execute_callback[default_queue] PASSED [ 68%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_process_workloads_routes_execute_callback[callback_queue] PASSED [ 70%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_execute_workload_runs_execute_task_before_airflow_3_3 SKIPPED [ 72%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_execute_workload_runs_base_executor_workload_on_airflow_3_3_plus PASSED [ 74%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::test_execute_workload_ignores_already_running_task PASSED [ 75%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_amqps_broker_url_builds_ssl_config PASSED [ 77%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_amqp_broker_url_still_builds_ssl_config PASSED [ 79%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_redis_mutual_tls_builds_ssl_config PASSED [ 81%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_amqps_mutual_tls_missing_key_cert_raises PASSED [ 83%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_ssl_active_without_cacert_uses_system_cas PASSED [ 85%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_amqps_broker_url_no_ssl_when_inactive PASSED [ 87%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_amqps_one_way_tls PASSED [ 88%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_redis_one_way_tls PASSED [ 90%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestAmqpsSslConfig::test_one_way_tls_ignores_key_cert PASSED [ 92%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCreateCeleryAppTeamIsolation::test_custom_celery_config_options_applied PASSED [ 94%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCreateCeleryAppTeamIsolation::test_default_celery_config_options_skipped_via_identity_check PASSED [ 96%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCreateCeleryAppTeamIsolation::test_team_specific_broker_not_overwritten PASSED [ 98%]
providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCreateCeleryAppTeamIsolation::test_team_app_name_includes_team_name PASSED [100%]
=========================== short test summary info ============================
SKIPPED [1] providers/celery/tests/unit/celery/executors/test_celery_executor.py:157: Test only for Airflow < 3.2
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_exception_propagation' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [3] providers/celery/tests/unit/celery/executors/test_celery_executor.py:219: Airflow 3 doesn't have execute_command anymore
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_task_instances_none' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_task_instances' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_try_adopt_with_and_without_external_executor_id' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_cleanup_stuck_queued_tasks' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:702: The test 'providers/celery/tests/unit/celery/executors/test_celery_executor.py::TestCeleryExecutor::test_revoke_task' requires one of 'mysql', 'postgres' backend started and 'BACKEND' environment variable to be set (currently it set to None). It can be set by specifying backend at breeze startup.
SKIPPED [1] devel-common/src/tests_common/pytest_plugin.py:661: The test is skipped because it has quarantined marker. And --include-quarantined flag is not passed to pytest. <Function test_send_workloads_to_celery_hang>
SKIPPED [1] providers/celery/tests/unit/celery/executors/test_celery_executor.py:997: pre-3.3 compatibility path only applies before Airflow 3.3
================== 42 passed, 12 skipped, 1 warning in 3.78s ===================\n- All checks passed!\n- 2 files already formatted
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "lohyenshen (via GitHub)" <gi...@apache.org>.
lohyenshen commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4956580988
hi folks, which helm-chart version is this fix published to?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "ilyaberd (via GitHub)" <gi...@apache.org>.
ilyaberd commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-5172851824
> hi folks, may I know which helm-chart version / apache-airflow-providers-celery version is this fix published to?
apache-airflow-providers-celery==3.21.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "anmolxlight (via GitHub)" <gi...@apache.org>.
anmolxlight commented on code in PR #67127:
URL: https://github.com/apache/airflow/pull/67127#discussion_r3475737649
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -389,24 +408,12 @@ def send_workload_to_executor(
Send workload to executor (serialized and executed as a Celery task).
This function is called in ProcessPoolExecutor subprocesses. To avoid pickling issues with
- team-specific Celery apps, we pass the team_name and reconstruct the Celery app here.
+ team-specific Celery apps, we pass the team_name and create the app in the subprocess. The app
+ is cached per subprocess and team so Celery backend resolution is amortized across publishes.
Review Comment:
Addressed in f4d1080194 — tightened docstring to clarify the cache's durable win is the scheduler-inline path, not the ProcessPoolExecutor subprocess path.
##########
providers/celery/tests/unit/celery/executors/test_celery_executor.py:
##########
@@ -551,6 +558,100 @@ def test_send_workload_uses_external_executor_id_as_celery_task_id():
assert result.task_id == pre_assigned_id
+@pytest.mark.parametrize("team_name", [None, "team-a"])
+def test_get_celery_app_for_workload_reuses_cache_for_same_team(team_name):
+ first_app = mock.Mock()
+ second_app = mock.Mock()
+
+ with mock.patch(
+ "airflow.providers.celery.executors.celery_executor_utils.create_celery_app",
+ side_effect=[first_app, second_app],
+ ) as mock_create_celery_app:
+ assert celery_executor_utils._get_celery_app_for_workload(team_name) is first_app
+ assert celery_executor_utils._get_celery_app_for_workload(team_name) is first_app
+
+ mock_create_celery_app.assert_called_once()
+
+
+def test_get_celery_app_for_workload_keeps_cache_team_scoped():
+ team_a_app = mock.Mock()
+ team_b_app = mock.Mock()
+
+ with mock.patch(
+ "airflow.providers.celery.executors.celery_executor_utils.create_celery_app",
+ side_effect=[team_a_app, team_b_app],
+ ) as mock_create_celery_app:
+ assert celery_executor_utils._get_celery_app_for_workload("team-a") is team_a_app
+ assert celery_executor_utils._get_celery_app_for_workload("team-b") is team_b_app
+
+ assert mock_create_celery_app.call_count == 2
+
+
+def test_send_workload_reuses_celery_app_for_same_team():
+ """Publishing multiple workloads for the same team reuses the cached Celery app."""
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
Review Comment:
Addressed in f4d1080194 — removed the two redundant cache_clear() calls from the unit tests; the autouse fixture already handles that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "seanmuth (via GitHub)" <gi...@apache.org>.
seanmuth commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4481473812
Drive-by from #67123 — confirming this matches the fix I proposed there. Both shapes (`@lru_cache` on a thin wrapper keyed on `team_name`) are functionally equivalent; preserves AIP-67 isolation while restoring per-subprocess amortization of the result-backend resolution that pre-3.16.0 was getting from the module-level `app` singleton.
Sharing some empirical numbers I gathered while reproducing #67123, in case they're useful for the reviewer:
**Per-publish overhead, measured as `create_celery_app(conf); _ = app.backend` (the path inside `[celery] operation_timeout`).** Astro Runtime 13.4.0 (Airflow 2.11.0), 1 vCPU / 2 GiB scheduler, 656 installed distributions, 100 iterations.
| Configuration | min | p50 | p95 | p99 | max | iter/s |
|---|---:|---:|---:|---:|---:|---:|
| Idle, no load | 48ms | 51ms | 67ms | 82ms | 86ms | 18.8 |
| Idle + 1.5 GiB anon allocation | 48 | 52 | 89 | 95 | 100 | 17.5 |
| 5 modest Dags with mapped tasks running | 52 | 89 | 232 | 363 | 385 | 9.6 |
| Same Dag load + 1.0 GiB anon allocation | 51 | 112 | 498 | 698 | 718 | 5.7 |
Reference: a real production deployment with the regression (~682 distributions, ~640 active Dags, periodic scheduler OOMs) shows `max=558ms` with consistent `Task Timeout Error for Task` log entries firing at the default 1.0s `operation_timeout`. Downgrading to providers-celery 3.15.x in the same image collapses the per-publish path to microseconds (singleton's backend is already resolved), confirming the regression's scope.
The takeaway for sizing the fix's expected impact: with this cache in place, the per-subprocess first-publish cost is paid once per team_name and then amortized over the subprocess's lifetime, restoring the pre-3.16.0 behavior. `apply_async` afterwards is just the broker round-trip plus a cached attribute access.
A couple of small things you might want to consider rolling in (entirely optional — happy if you ignore):
1. A docstring on `_get_celery_app_for_workload` linking to the issue and briefly noting *why* the cache exists — future readers wondering about the indirection will appreciate the breadcrumb. Without context, "subprocess-local Celery app cached by team name for task publishing" reads like incidental optimization rather than a regression fix.
2. A third unit test that exercises `_get_celery_app_for_workload` directly (without going through `send_workload_to_executor`) — asserting that consecutive calls for the same team return the same object, and that different team names return distinct objects. The two existing tests cover the end-to-end behavior well; a direct-on-the-cache test would catch refactors that move work out of `_get_celery_app_for_workload` without breaking the integration path.
Comparison point in case it's useful: PR #61798 implemented the same AIP-67 multi-team feature for `KubernetesExecutor` without the per-publish cost because that executor's design naturally holds the team-aware `KubeConfig` once at `__init__`. Worth noting that the per-publish reconstruction wasn't an inherent AIP-67 requirement — it was an artifact of the `ProcessPoolExecutor.submit(...)` shape the Celery publisher pool already had. This PR effectively brings the Celery side back in line with that pattern.
Either way, the fix is correct and I'd love to see it merged. Thanks for picking this up so quickly.
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @seanmuth before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "anmolxlight (via GitHub)" <gi...@apache.org>.
anmolxlight commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4800343965
Hey @kaxil — all 4 review threads are now resolved:
1. → : no arbitrary limit
2. Docstring updated to cover both scheduler-inline & subprocess paths
3. leftover removed from
4. parametrized into the cache-reuse test
Could you give it another look when you get a chance? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #67127:
URL: https://github.com/apache/airflow/pull/67127
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "anmolxlight (via GitHub)" <gi...@apache.org>.
anmolxlight commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4483883682
Rolled in the optional follow-ups: documented why the cache exists and where it lives, added direct cache coverage including `team_name=None`, removed the leftover type-only block, and changed the cache to avoid an arbitrary team limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4529572385
@anmolxlight A few things need addressing before review — see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **CI fails**: `Integration and System Tests / Integration: providers celery` (and possibly other checks — see the [Checks tab](https://github.com/apache/airflow/pull/67127/checks) for the full list).
No rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
---
Drafted-by: Claude Code (Opus 4.7); reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "anmolxlight (via GitHub)" <gi...@apache.org>.
anmolxlight commented on code in PR #67127:
URL: https://github.com/apache/airflow/pull/67127#discussion_r3263291970
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
Review Comment:
Changed. Replaced the arbitrary `maxsize=8` with `@cache`, so configured teams are cached without eviction in the publishing process.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
+def _get_celery_app_for_workload(team_name: str | None) -> Celery:
+ """Return a subprocess-local Celery app cached by team name for task publishing."""
Review Comment:
Fixed. Updated the docstring to describe both scheduler-inline and publisher-subprocess execution paths, and made the cache process-local rather than subprocess-specific.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
+def _get_celery_app_for_workload(team_name: str | None) -> Celery:
+ """Return a subprocess-local Celery app cached by team name for task publishing."""
+ if TYPE_CHECKING:
+ _conf: ExecutorConf | AirflowConfigParser
Review Comment:
Fixed. Removed the leftover `TYPE_CHECKING` annotation block from `_get_celery_app_for_workload`.
##########
providers/celery/tests/unit/celery/executors/test_celery_executor.py:
##########
@@ -77,6 +77,13 @@
pytestmark = pytest.mark.db_test
+@pytest.fixture(autouse=True)
+def clear_cached_workload_celery_apps():
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
+ yield
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
Review Comment:
Fixed. Added direct cache tests for `team_name=None`, same-team reuse, and distinct team names.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4798268504
@anmolxlight — this PR has 4 unresolved review thread(s) that still look like they need your attention. Once you've addressed them (push changes and/or reply in-thread), please resolve the threads or reply to confirm, and give the reviewer a nudge for another look. Thanks!
See the [PR quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
<sub>Automated first-pass triage note drafted by an AI-assisted tool — may get things wrong; once addressed, a real Apache Airflow maintainer takes the next look. ([why automated](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md))</sub>
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "anmolxlight (via GitHub)" <gi...@apache.org>.
anmolxlight commented on PR #67127:
URL: https://github.com/apache/airflow/pull/67127#issuecomment-4800347394
Hey @kaxil — all 4 review threads are now resolved:
1. `maxsize=8` -> `@cache`: no arbitrary limit
2. Docstring updated to cover both scheduler-inline & subprocess paths
3. `TYPE_CHECKING` leftover removed from `_get_celery_app_for_workload`
4. `team_name=None` parametrized into the cache-reuse test
Could you give it another look when you get a chance? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #67127:
URL: https://github.com/apache/airflow/pull/67127#discussion_r3262851195
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
+def _get_celery_app_for_workload(team_name: str | None) -> Celery:
+ """Return a subprocess-local Celery app cached by team name for task publishing."""
Review Comment:
The "subprocess-local" claim only holds when `_send_workloads_to_celery` actually uses the `ProcessPoolExecutor` branch. In `celery_executor.py:243-245`, the single-workload (or `sync_parallelism=1`) path runs `send_workload_to_executor` inline via `map(...)` in the scheduler process. In that case this cache lives in the scheduler itself and keeps the cached Celery apps' broker connections open there too. Worth either updating the docstring to reflect "scheduler process or publisher subprocess, depending on path", or being explicit that this is intentional.
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
+def _get_celery_app_for_workload(team_name: str | None) -> Celery:
+ """Return a subprocess-local Celery app cached by team name for task publishing."""
+ if TYPE_CHECKING:
+ _conf: ExecutorConf | AirflowConfigParser
Review Comment:
This `if TYPE_CHECKING: _conf: ExecutorConf | AirflowConfigParser` is a copy-paste leftover from the old `send_workload_to_executor`. The annotation never escapes this function and `_conf` already gets a concrete type from the if/else assignment below, so this whole two-line block can be dropped.
##########
providers/celery/tests/unit/celery/executors/test_celery_executor.py:
##########
@@ -77,6 +77,13 @@
pytestmark = pytest.mark.db_test
+@pytest.fixture(autouse=True)
+def clear_cached_workload_celery_apps():
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
+ yield
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
Review Comment:
The two new tests cover `team-a` reuse and `team-a` vs `team-b` separation, but not `team_name=None`, which is the path the vast majority of deployments hit (no multi-team config). A third parametrized case asserting `None` also hits the cache would catch a regression where `lru_cache` started treating `None` specially.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #67127:
URL: https://github.com/apache/airflow/pull/67127#discussion_r3262820839
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -160,6 +160,21 @@ def create_celery_app(team_conf: ExecutorConf | AirflowConfigParser) -> Celery:
return celery_app
+@lru_cache(maxsize=8)
Review Comment:
Why 8?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Cache Celery apps when publishing workloads [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #67127:
URL: https://github.com/apache/airflow/pull/67127#discussion_r3475540805
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -389,24 +408,12 @@ def send_workload_to_executor(
Send workload to executor (serialized and executed as a Celery task).
This function is called in ProcessPoolExecutor subprocesses. To avoid pickling issues with
- team-specific Celery apps, we pass the team_name and reconstruct the Celery app here.
+ team-specific Celery apps, we pass the team_name and create the app in the subprocess. The app
+ is cached per subprocess and team so Celery backend resolution is amortized across publishes.
Review Comment:
This docstring says the app cache amortizes backend resolution "across publishes", but the `ProcessPoolExecutor` in `_send_workloads_to_celery` is recreated and torn down on every publish, so in the subprocess path the cache only lives for one batch. The durable win is the inline path (`len==1` or `sync_parallelism==1`), which runs in the long-lived scheduler process and is exactly the 1-vCPU repro from #67123. Might be worth tightening the wording so a future reader doesn't assume the pool-path cache survives across publishes.
##########
providers/celery/tests/unit/celery/executors/test_celery_executor.py:
##########
@@ -551,6 +558,100 @@ def test_send_workload_uses_external_executor_id_as_celery_task_id():
assert result.task_id == pre_assigned_id
+@pytest.mark.parametrize("team_name", [None, "team-a"])
+def test_get_celery_app_for_workload_reuses_cache_for_same_team(team_name):
+ first_app = mock.Mock()
+ second_app = mock.Mock()
+
+ with mock.patch(
+ "airflow.providers.celery.executors.celery_executor_utils.create_celery_app",
+ side_effect=[first_app, second_app],
+ ) as mock_create_celery_app:
+ assert celery_executor_utils._get_celery_app_for_workload(team_name) is first_app
+ assert celery_executor_utils._get_celery_app_for_workload(team_name) is first_app
+
+ mock_create_celery_app.assert_called_once()
+
+
+def test_get_celery_app_for_workload_keeps_cache_team_scoped():
+ team_a_app = mock.Mock()
+ team_b_app = mock.Mock()
+
+ with mock.patch(
+ "airflow.providers.celery.executors.celery_executor_utils.create_celery_app",
+ side_effect=[team_a_app, team_b_app],
+ ) as mock_create_celery_app:
+ assert celery_executor_utils._get_celery_app_for_workload("team-a") is team_a_app
+ assert celery_executor_utils._get_celery_app_for_workload("team-b") is team_b_app
+
+ assert mock_create_celery_app.call_count == 2
+
+
+def test_send_workload_reuses_celery_app_for_same_team():
+ """Publishing multiple workloads for the same team reuses the cached Celery app."""
+ celery_executor_utils._get_celery_app_for_workload.cache_clear()
Review Comment:
The autouse `clear_cached_workload_celery_apps` fixture already clears the cache before and after every test in this module, so this explicit `cache_clear()` (and the matching one in `test_send_workload_keeps_celery_app_cache_team_scoped`) is redundant. The other two new cache tests rely on the fixture. (The integration-test `cache_clear()` calls are necessary there, since that file has no such fixture.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org