Issue 1:
Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

While using the Dynamic Task Mapping the following error is thrown in worker while loading DAG.

Setting task policy for Dag: DAG1 Task: DownstreamDagTrigger

[2023-01-09 09:44:36,545: ERROR/ForkPoolWorker-7] can’t set attribute

Traceback (most recent call last):

File “/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py”, line 523, in collect_dags

found_dags = self.process_file(filepath, only_if_updated=only_if_updated, safe_mode=safe_mode)

File “/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py”, line 291, in process_file

found_dags = self._process_modules(filepath, mods, file_last_changed_on_disk)

File “/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py”, line 407, in _process_modules

self.bag_dag(dag=dag, root_dag=dag)

File “/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py”, line 434, in bag_dag

self._bag_dag(dag=dag, root_dag=root_dag, recursive=True)

File “/home/airflow/.local/lib/python3.8/site-packages/airflow/models/dagbag.py”, line 451, in _bag_dag

settings.task_policy(task)

File “/home/airflow/.local/lib/python3.8/site-packages/datahub_provider/_plugin.py”, line 308, in custom_task_policy

task_policy(task)

File “/home/airflow/.local/lib/python3.8/site-packages/datahub_provider/_plugin.py”, line 297, in task_policy

task.on_failure_callback = _wrap_on_failure_callback(task.on_failure_callback)

AttributeError: can’t set attribute

This error is thrown can some please help.

I am using the TriggerDagRunOperator for Triggering Multiple Downstream DAG.

What you think should happen instead

The DAG Dependency Graph should be shown like

How to reproduce

You can use

TriggerDagRunOperator.partial(

task_id="Task1",

dag=dag,

reset_dag_run=True,

).expand(trigger_dag_id=task_conf.ops_params['downstream_dag'])

Operating System

Deployed on EKS

Versions of Apache Airflow Providers

apache-airflow 2.3.4

apache-airflow-providers-amazon 5.0.0

apache-airflow-providers-apache-hive 4.1.0

apache-airflow-providers-apache-livy 3.1.0

apache-airflow-providers-apache-spark 4.0.0

apache-airflow-providers-celery 3.0.0

apache-airflow-providers-cncf-kubernetes 4.3.0

apache-airflow-providers-common-sql 1.3.1

apache-airflow-providers-docker 3.1.0

apache-airflow-providers-elasticsearch 4.2.0

apache-airflow-providers-ftp 3.1.0

apache-airflow-providers-google 8.3.0

apache-airflow-providers-grpc 3.0.0

apache-airflow-providers-hashicorp 3.1.0

apache-airflow-providers-http 4.0.0

apache-airflow-providers-imap 3.0.0

apache-airflow-providers-microsoft-azure 4.2.0

apache-airflow-providers-mysql 3.2.0

apache-airflow-providers-odbc 3.1.1

apache-airflow-providers-postgres 5.2.0

apache-airflow-providers-redis 3.0.0

apache-airflow-providers-sendgrid 3.0.0

apache-airflow-providers-sftp 4.0.0

apache-airflow-providers-slack 5.1.0

apache-airflow-providers-sqlite 3.2.0

apache-airflow-providers-ssh 3.1.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

No response

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

Issue 2:
Apache Airflow version

2.3.2 (latest released)

What happened

Using cluster policies, I set a default on_failure_callback.

This works great with all tasks expect with task new task mapping.

What you think should happen instead

Policies should work on all tasks, including the dynamic mapping or generate an exception with the error.

How to reproduce

Overwrite any parameters using the task_policy.

Operating System

Official docker image

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

No response

Anything else

No response

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