You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "deepak4babu (via GitHub)" <gi...@apache.org> on 2025/02/13 07:43:39 UTC
[I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
deepak4babu opened a new issue, #46714:
URL: https://github.com/apache/airflow/issues/46714
### Apache Airflow version
Other Airflow 2 version (please specify below)
### If "Other Airflow 2 version" selected, which one?
2.10.3
### What happened?
We are using Celery kubernetes executor in our airflow setup. Based on different task parameters, we are trying to add executor_config using airflow policy (task_policy). The policy is failing for dynamic task alone with below error
```
property 'executor_config' of 'DecoratedMappedOperator' object has no setter
```
### What you think should happen instead?
Airflow policy should be able to configure executor_config to dynamic task without any issues.
### How to reproduce
Add the below task_policy in airflow policy
```python
def task_policy(task: BaseOperator) -> None:
task.queue = "kubernetes"
task.executor_config = {
"pod_override": k8s.V1Pod(
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
name="base",
resources=k8s.V1ResourceRequirements(
requests={
"cpu": "500m",
},
),
)
]
)
)
}
```
### Operating System
PRETTY_NAME="Debian GNU/Linux 12 (bookworm)" NAME="Debian GNU/Linux" VERSION_ID="12" VERSION="12 (bookworm)" VERSION_CODENAME=bookworm ID=debian HOME_URL="https://www.debian.org/" SUPPORT_URL="https://www.debian.org/support" BUG_REPORT_URL="https://bugs.debian.org/"
### Versions of Apache Airflow Providers
apache-airflow-providers-celery==3.8.3
apache-airflow-providers-cncf-kubernetes==10.1.0
apache-airflow-providers-common-compat==1.2.1
apache-airflow-providers-common-io==1.4.2
apache-airflow-providers-common-sql==1.19.0
apache-airflow-providers-fab==1.5.0
apache-airflow-providers-ftp==3.11.1
apache-airflow-providers-http==4.13.2
apache-airflow-providers-imap==3.7.0
apache-airflow-providers-smtp==1.8.0
apache-airflow-providers-sqlite==3.9.0
### Deployment
Other
### Deployment details
We deployed using our custom helm chart for airflow.
### Anything else?
Error message
```
[2025-02-13T06:07:29.774+0000] {dagbag.py:532} ERROR - property 'executor_config' of 'DecoratedMappedOperator' object has no setter
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 528, in _bag_dag
settings.task_policy(task)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/settings.py", line 164, in task_policy
return POLICY_PLUGIN_MANAGER.hook.task_policy(task=task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_hooks.py", line 513, in __call__
return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
raise exception.with_traceback(exception.__traceback__)
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_callers.py", line 103, in _multicall
res = hook_impl.function(*args)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/<>/airflow/airflow_local_settings.py", line 285, in task_policy
task.executor_config = {
^^^^^^^^^^^^^^^^^^^^
AttributeError: property 'executor_config' of 'DecoratedMappedOperator' object has no setter
[2025-02-13T06:07:29.776+0000] {dagbag.py:484} ERROR - Failed to bag_dag: /opt/airflow/dags/<dag_name>.py
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 528, in _bag_dag
settings.task_policy(task)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/settings.py", line 164, in task_policy
return POLICY_PLUGIN_MANAGER.hook.task_policy(task=task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_hooks.py", line 513, in __call__
return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
raise exception.with_traceback(exception.__traceback__)
File "/home/airflow/.local/lib/python3.12/site-packages/pluggy/_callers.py", line 103, in _multicall
res = hook_impl.function(*args)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/<>/airflow/airflow_local_settings.py", line 285, in task_policy
task.executor_config = {
^^^^^^^^^^^^^^^^^^^^
AttributeError: property 'executor_config' of 'DecoratedMappedOperator' object has no setter
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 480, in _process_modules
self.bag_dag(dag=dag, root_dag=dag)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 499, in bag_dag
self._bag_dag(dag=dag, root_dag=root_dag, recursive=True)
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 533, in _bag_dag
raise AirflowClusterPolicyError(e)
airflow.exceptions.AirflowClusterPolicyError: property 'executor_config' of 'DecoratedMappedOperator' object has no setter
```
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on issue #46714:
URL: https://github.com/apache/airflow/issues/46714#issuecomment-2738019101
You are experiencing this issue because the `executor_config` property for mapped tasks does not have a defined setter. This has been reported before in [#24547](https://github.com/apache/airflow/issues/24547) and [#37013](https://github.com/apache/airflow/issues/37013). PRs [#28313](https://github.com/apache/airflow/pull/28313) and [#37828](https://github.com/apache/airflow/pull/37828) fixed this issue for some properties by defining setters for them. However, the `executor_config` property does not have a setter defined yet.
I also experienced this issue when I was trying to implement a task policy for `executor_config` and was able to work around this by explicitly overriding `executor_config` for the `MappedOperator` and `DecoratedMappedOperator`in `partial_kwargs`.
```
from airflow.policies import hookimpl
from kubernetes.client import models as k8s
@hookimpl
def task_policy(task: "BaseOperator"):
ke_pod_default = {
"pod_override": k8s.V1Pod(
spec=k8s.V1PodSpec(
containers=[
k8s.V1Container(
name="base",
resources=k8s.V1ResourceRequirements(
requests={"cpu": 1, "memory": "2048Mi", "ephemeral-storage": "1Gi"},
limits={"cpu": 1, "memory": "2048Mi", "ephemeral-storage": "1Gi"}
)
)
]
)
)
}
if task.__class__.__name__ in ["MappedOperator", "DecoratedMappedOperator"]:
if task.partial_kwargs["executor_config"] == {}:
task.partial_kwargs["executor_config"] = ke_pod_default
else:
if task.executor_config == {}:
task.executor_config = ke_pod_default
```
I recommend implementing this workaround until we are able to add a setter for the `executor_config` property.
--
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: [I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed issue #46714: Unable to assign executor_config for dynamic tasks using airflow policy
URL: https://github.com/apache/airflow/issues/46714
--
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: [I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #46714:
URL: https://github.com/apache/airflow/issues/46714#issuecomment-5324885176
This issue has been closed because it has not received response from the issue author.
--
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: [I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
Posted by "karenbraganz (via GitHub)" <gi...@apache.org>.
karenbraganz commented on issue #46714:
URL: https://github.com/apache/airflow/issues/46714#issuecomment-2738333414
Another more straightforward option is to override pod resources with a [pod policy](https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/cluster-policies.html#airflow.policies.pod_mutation_hook) instead of a task policy like in [this example](https://www.astronomer.io/docs/learn/airflow-cluster-policies/#example-3).
I was only able to use a task policy for my use case because I only wanted to override `executor_config` if it wasn't already set at the task level. If you want to set specific resources for all pods, you could try the pod policy.
Still, in my opinion, it would be useful to add a setter to `executor_config` so it can be overridden for specific tasks using task policies.
--
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: [I] Unable to assign executor_config for dynamic tasks using airflow policy [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #46714:
URL: https://github.com/apache/airflow/issues/46714#issuecomment-5010418739
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.
--
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