Modifying the task attribute for DAG with dynamic task mapping didn't work

[Bowrna]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

I have a cluster policy enabled where i add the fields for task in it. This works fine for all the task. I tried to add a DAG with dynamic task mapping enabled and it didn't load in the UI. On checking the scheduler logs, i found the below log trace.

[2023-03-10T00:01:24.998+0000] {processor.py:154} INFO - Started process (PID=498) to work on /root/airflow/dags/test_dynamic_task_mapping.py
[2023-03-10T00:01:25.000+0000] {processor.py:756} INFO - Processing file /root/airflow/dags/test_dynamic_task_mapping.py for tasks to queue
[2023-03-10T00:01:25.002+0000] {logging_mixin.py:120} INFO - [2023-03-10T00:01:25.001+0000] {dagbag.py:537} INFO - Filling up the DagBag from /root/airflow/dags/test_dynamic_task_mapping.py
[2023-03-10T00:01:25.042+0000] {logging_mixin.py:120} INFO - No success callback added. So adding here
[2023-03-10T00:01:25.044+0000] {logging_mixin.py:120} INFO - [2023-03-10T00:01:25.042+0000] {dagbag.py:564} ERROR - can't set attribute
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 551, in collect_dags
    found_dags = self.process_file(filepath, only_if_updated=only_if_updated, safe_mode=safe_mode)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 307, in process_file
    found_dags = self._process_modules(filepath, mods, file_last_changed_on_disk)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 437, in _process_modules
    self.bag_dag(dag=dag, root_dag=dag)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 464, in bag_dag
    self._bag_dag(dag=dag, root_dag=root_dag, recursive=True)
  File "/usr/local/lib/python3.8/site-packages/airflow/models/dagbag.py", line 481, in _bag_dag
    settings.task_policy(task)
  File "/root/airflow/config/airflow_local_settings.py", line 270, in task_policy
    _check_task_rules(task)
  File "/root/airflow/config/airflow_local_settings.py", line 258, in _check_task_rules
    rule(current_task)
  File "/root/airflow/config/airflow_local_settings.py", line 221, in add_success_callback
    task.on_success_callback= task_success_alert
AttributeError: can't set attribute
[2023-03-10T00:01:25.045+0000] {processor.py:768} WARNING - No viable dags retrieved from /root/airflow/dags/test_dynamic_task_mapping.py
[2023-03-10T00:01:25.058+0000] {processor.py:176} INFO - Processing /root/airflow/dags/test_dynamic_task_mapping.py took 0.074 seconds
[2023-03-10T00:02:06.276+0000] {processor.py:154} INFO - Started process (PID=1524) to work on /root/airflow/dags/test_dynamic_task_mapping.py

What you think should happen instead

The attribute setting for task in cluster policy should not throw error

How to reproduce

I have added a cluster policy setting like mentioned here
https://gist.github.com/Bowrna/1994894beea39fa8e1c269591b7f0346
And then i tried to load the example dynamic task mapping DAG that comes as loaded example DAG in the UI

Operating System

centos

Versions of Apache Airflow Providers

Version : 2.4.2

Deployment

Virtualenv installation

Deployment details

No response

Anything else

I would love to contribute in fixing the issue. But i would like to get the views from the community members more about this issue.

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

[hussein-awala]

Indeed setting MappedOperator properties was not supported in 2.4.2, but we fixed that in #28313 which is available in 2.5.1.

I just tested setting the on_success_callback with Airflow 2.5.1 and it worked as expected, can you try to upgrade to the latest version and test your cluster policy?

[Bowrna]

thanks @hussein-awala for the update. this helps a lot