You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "o-nikolas (via GitHub)" <gi...@apache.org> on 2026/01/17 01:01:30 UTC
[PR] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
o-nikolas opened a new pull request, #60675:
URL: https://github.com/apache/airflow/pull/60675
Updating the Celery executor to read from team based config and also support multiple instances running concurrently.
The latter is the largest source of changes. Much of the celery configuration (both Airflow config and Celery config) was module based. Modules are cached and shared in Python. So the majority of the changes are moving that module level config code to be function based (while trying to also maintain backwards compatibility).
The way Celery tasks are sent to workers also changed as a consequence of this. 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).
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2729131365
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
What were your steps for this reproduction by the way? I'd like to test the same way you did
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2713195448
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -102,6 +104,38 @@ def _get_celery_app() -> Celery:
return Celery(celery_app_name, config_source=get_celery_configuration())
+def create_celery_app(team_conf) -> Celery:
Review Comment:
Do we know the type? If yes, it would be better to have type annotation
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor.py:
##########
@@ -206,6 +217,8 @@ def _send_tasks_to_celery(self, task_tuples_to_send: Sequence[TaskInstanceInCele
chunksize = self._num_tasks_per_send_process(len(task_tuples_to_send))
num_processes = min(len(task_tuples_to_send), self._sync_parallelism)
+ # Use ProcessPoolExecutor with team_name instead of task objects to avoid pickling issues.
Review Comment:
I do not see it being used?
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2743923583
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
I've made the fixes and added the unit test. Can you remove the request for changes and do another review @dheerajturaga ? 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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2744596079
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,32 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
+ raise AirflowConfigException(
Review Comment:
Can we raise SystemExit instead, that way we are aligned with the other SystemExit exceptions thrown for other missing arguments? Current approach throws a stack trace which can be confusing to the user.
```suggestion
raise SystemExit(
```
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2724383833
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,27 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ try:
+ from airflow.executors.base_executor import ExecutorConf
+ except ImportError:
+ raise SystemExit(
+ f"Error: Multi-team support (--team {args.team}) requires Airflow >=3.2. "
+ "Please upgrade Airflow or remove the --team parameter."
+ )
Review Comment:
Add attribute check before creating the ExecutorConf object?
```suggestion
)
# Check if ExecutorConf has required methods for multi-team support
if not (hasattr(ExecutorConf, 'has_option') and hasattr(ExecutorConf, 'getsection')):
raise SystemExit(
f"Error: Multi-team support (--team {args.team}) requires Airflow >=3.2. "
"Please upgrade Airflow or remove the --team parameter."
)
```
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
@o-nikolas, this is failing backward compatibility. In Airflow 3.1.3, ExecutorConf exists but lacks critical methods like has_option, getsection, etc. were added in commit 5ba75af409 for Airflow 3.2+
```
if autoscale is None and config.has_option("celery", "worker_autoscale"):
AttributeError: 'ExecutorConf' object has no attribute 'has_option'
```
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2729408062
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
this is what I'm using for these backward compatibility tests:
Build the local airflow wheel:
`breeze release-management prepare-provider-distributions --skip-tag-check celery`
Start specific airflow version:
`breeze start-airflow -d -e --user-airflow-version 3.1.3 --executor CeleryExecutor --airflow-extras celery`
then within the container, shutdown the airflow celery worker. Then `pip install ./dist/<celery>.whl`
then launch the worker with `airflow celery worker --team my_team`
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2729130277
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
Thanks for testing! I had seen the same failure already and added checks in other places in the code for 3.2+ and/or attribute checks on ExecutorConf. But apparently some are missed in our unit/integ tests? Or we don't run that version on CI. I'll patch these up!
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #60675:
URL: https://github.com/apache/airflow/pull/60675#issuecomment-3765770027
@o-nikolas I have tested the general functionality of celery worker with your changes as well as tested out the CLI. Things are working as expected. However, I don't see how I can test the `--team` option. Seems like if I set a team the jobs still seem to be launching on the worker. How do you propose I test the multi team support here?
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2744588631
##########
providers/celery/src/airflow/providers/celery/cli/definition.py:
##########
@@ -101,6 +101,10 @@
help="Don't subscribe to other workers events",
action="store_true",
)
+ARG_TEAM = Arg(
+ ("-t", "--team"),
+ help="Team name for team-specific multi-team configuration (requires Airflow 3.1+)",
Review Comment:
```suggestion
help="Team name for team-specific multi-team configuration (requires Airflow 3.2+)",
```
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,32 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
+ raise AirflowConfigException(
Review Comment:
Can we raise SystemExit instead to be consistent with the other exceptions? Current approach throws a stacktrace which can get confusing to the user
```suggestion
raise SystemExit(
```
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,32 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
+ raise AirflowConfigException(
+ "Multi-team Celery workers require Airflow version 3.2 or higher. "
+ "Please upgrade your Airflow installation or remove the --team argument."
+ )
+ if not conf.getboolean("core", "multi_team", fallback=False):
+ raise AirflowConfigException(
Review Comment:
Same here
```suggestion
raise SystemExit(
```
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,32 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
+ raise AirflowConfigException(
+ "Multi-team Celery workers require Airflow version 3.2 or higher. "
+ "Please upgrade your Airflow installation or remove the --team argument."
+ )
+ if not conf.getboolean("core", "multi_team", fallback=False):
+ raise AirflowConfigException(
+ "Multi-team Celery workers require core.multi_team configuration to be enabled. "
Review Comment:
```suggestion
"Error: Multi-team Celery workers require core.multi_team configuration to be enabled. "
```
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,32 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
+ raise AirflowConfigException(
+ "Multi-team Celery workers require Airflow version 3.2 or higher. "
Review Comment:
```suggestion
"Error: Multi-team Celery workers require Airflow version 3.2 or higher. "
```
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #60675:
URL: https://github.com/apache/airflow/pull/60675#issuecomment-3770661646
> @o-nikolas I have tested the general functionality of celery worker with your changes as well as tested out the CLI. Things are working as expected. However, I don't see how I can test the `--team` option. Seems like if I set a team the jobs still seem to be launching on the worker. How do you propose I test the multi team support here?
Thanks for the thorough review @dheerajturaga! I've addressed those issues (slightly differently for one than your suggested patch). I'm currently struggling with back compat tests. It's slow/difficult because those tests to do not run successfully in breeze on my laptop. So I have to push to the PR to test each change.
As far as testing with the --team flag. For this you have to have a full multi-team setup, which we don't have great documentation for yet (coming soon). The most helpful testing is actually on the backcompat side (testing with airflow 2.11 and 3.1.X)
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2713772911
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor.py:
##########
@@ -206,6 +217,8 @@ def _send_tasks_to_celery(self, task_tuples_to_send: Sequence[TaskInstanceInCele
chunksize = self._num_tasks_per_send_process(len(task_tuples_to_send))
num_processes = min(len(task_tuples_to_send), self._sync_parallelism)
+ # Use ProcessPoolExecutor with team_name instead of task objects to avoid pickling issues.
Review Comment:
The process pool is below and the team name is plumbed through the `task_tuples_to_send` object
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga merged PR #60675:
URL: https://github.com/apache/airflow/pull/60675
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #60675:
URL: https://github.com/apache/airflow/pull/60675#issuecomment-3829831574
> Awesome! Thanks so much for your patience, I know there was a lot of back and forth. Current changes look good!
No worries @dheerajturaga! I appreciate the thorough review and testing :) I made sure to commit one of your suggestions to get you tagged as a co-author for your efforts!
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2724383833
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -189,8 +189,27 @@ def filter(self, record):
@_providers_configuration_loaded
def worker(args):
"""Start Airflow Celery worker."""
- # This needs to be imported locally to not trigger Providers Manager initialization
- from airflow.providers.celery.executors.celery_executor import app as celery_app
+ team_config = None
+ if hasattr(args, "team") and args.team:
+ # Multi-team is enabled, create team-specific Celery app and use team based config
+ try:
+ from airflow.executors.base_executor import ExecutorConf
+ except ImportError:
+ raise SystemExit(
+ f"Error: Multi-team support (--team {args.team}) requires Airflow >=3.2. "
+ "Please upgrade Airflow or remove the --team parameter."
+ )
Review Comment:
Add attribute check before creating the ExecutorConf object?
```suggestion
)
# Check if ExecutorConf has required methods for multi-team support
if not hasattr(ExecutorConf, 'getsection'):
raise SystemExit(
f"Error: Multi-team support (--team {args.team}) requires Airflow >=3.2. "
"Please upgrade Airflow or remove the --team parameter."
)
```
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2729411288
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
We should probably add a backward compatibility test for this
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2738263398
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -192,13 +192,18 @@ def worker(args):
team_config = None
if hasattr(args, "team") and args.team:
# Multi-team is enabled, create team-specific Celery app and use team based config
- try:
- from airflow.executors.base_executor import ExecutorConf
- except ImportError:
- raise SystemExit(
- f"Error: Multi-team support (--team {args.team}) requires Airflow >=3.2. "
- "Please upgrade Airflow or remove the --team parameter."
+ # This requires Airflow 3.2+, and core.multi_team config to be true to be enabled.
+ if not AIRFLOW_V_3_2_PLUS:
Review Comment:
Nice, this should be much safer than the previous approach
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2742724345
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
It is a bit of an invalid use case, it requires the user to be trying to use a feature on a version of Airflow where the feature doesn't exist. But it may happen. I can look into writing a test to cover it.
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2743932450
##########
providers/celery/src/airflow/providers/celery/cli/celery_command.py:
##########
@@ -218,8 +237,8 @@ def worker(args):
autoscale = args.autoscale
skip_serve_logs = args.skip_serve_logs
- if autoscale is None and conf.has_option("celery", "worker_autoscale"):
- autoscale = conf.get("celery", "worker_autoscale")
+ if autoscale is None and config.has_option("celery", "worker_autoscale"):
Review Comment:
@o-nikolas, many users may not be aware of what airflow version they are on. It could happen that they may just read the documentation available online and try out a feature that probably doesn't exist. I removed the request for changes, I will do a review in a couple of hours! thanks for addressing these !
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2701247925
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -73,7 +73,7 @@
# the type as the union of both kinds
CommandType = Sequence[str]
- TaskInstanceInCelery: TypeAlias = tuple[TaskInstanceKey, workloads.All | CommandType, str | None, Task]
+ TaskInstanceInCelery: TypeAlias = tuple[TaskInstanceKey, workloads.All | CommandType, str | None, str]
Review Comment:
The fourth element is typed as str (non-optional), but self.team_name in the BaseExecutor class is typed as str | None. When the code creates tuples with self.team_name, it's passing a potentially None value where a non-optional str is expected.
```suggestion
TaskInstanceInCelery: TypeAlias = tuple[
TaskInstanceKey, workloads.All | CommandType, str | None, str | None
]
```
##########
providers/celery/tests/integration/celery/test_celery_executor.py:
##########
@@ -152,11 +162,24 @@ def teardown_method(self) -> None:
def test_celery_integration(self, broker_url, executor_config):
from airflow.providers.celery.executors import celery_executor, celery_executor_utils
- def fake_execute_workload(command):
- if "fail" in command:
- raise AirflowException("fail")
-
- with _prepare_app(broker_url, execute=fake_execute_workload) as app:
+ if AIRFLOW_V_3_0_PLUS:
+ # Airflow 3: execute_workload receives JSON string
+ def fake_execute(input: str):
+ """Fake execute_workload that parses JSON and fails for tasks with 'fail' in task_id."""
+ import json
+
+ workload_dict = json.loads(input)
+ # Check if this is a task that should fail (task_id contains "fail")
+ if "ti" in workload_dict and "task_id" in workload_dict["ti"]:
+ if "fail" in workload_dict["ti"]["task_id"]:
+ raise AirflowException("fail")
+ else:
+ # Airflow 2: execute_command receives command list
+ def fake_execute(command):
Review Comment:
Mypy requires all conditional function variants to have identical signatures
```suggestion
def fake_execute(command: str):
```
--
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] AIP-67 - Multi Team: Update Celery Executor to support multi team [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #60675:
URL: https://github.com/apache/airflow/pull/60675#discussion_r2713766073
##########
providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py:
##########
@@ -102,6 +104,38 @@ def _get_celery_app() -> Celery:
return Celery(celery_app_name, config_source=get_celery_configuration())
+def create_celery_app(team_conf) -> Celery:
Review Comment:
It's an ExecutorConf type, I'll update it.
--
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