Unable to assign executor_config for dynamic tasks using airflow policy

[deepak4babu]

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

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

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

[karenbraganz]

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 and #37013. PRs #28313 and #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 DecoratedMappedOperatorin 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.

[karenbraganz]

Another more straightforward option is to override pod resources with a pod policy instead of a task policy like in this example.

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.

[github-actions]

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.

[github-actions]

This issue has been closed because it has not received response from the issue author.