Description

We would like to be able to implement fail fast strategy of the Dag. It means if any of the task fails it is required to stop all the rest tasks running in parallel and mark the entire Dag as FAILED.
For example, we have 3 tasks in the first Task Group, which are executed in parallel. The second Task Group or a task needs not only to fail (which can be achieved by setting ALL_SUCCESS trigger rule) but also stop the running tasks.
Currently we implement it by adding one more nested Task Group into the first group with two tasks:

1. Task 1 with ONE_FAILED trigger rule is PythonOperator, which calls the function, which looks for the upstream tasks of this task in the internal TaskInstance Airflow table and mark them as FAILED. E.g.

@provide_session
def mark_tasks_as_fail(task, data_interval_start, dag, session=None, **_):
    upstream_task_instances = session.query(TaskInstance).filter(
        TaskInstance.dag_id == dag.dag_id,
        TaskInstance.execution_date == data_interval_start,
        TaskInstance.task_id.in_(task.upstream_task_ids)).all()
    for ti in upstream_task_instances:
        ti.state = TaskInstanceState.FAILED

2. Task 2 with ALL_DONE trigger rule is also PythonOperator with the function, which just raises an exception to cause the next task or group with ALL_SUCCESS trigger rule to fail. Eventually the entire Dag fails.

We do not like such approach, since it relies on the Airflow internals.

Please advise if there is a way to implement what is needed by using some high-level API.

Use case/motivation

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

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