Add support for automatic clearing of failed child DAG tasks in TriggerDagRunOperator

[Pad71]

Description

I propose an enhancement to TriggerDagRunOperator that would allow automatic clearing of failed task instances in a triggered DAG run if that DAG run previously failed.

Use case/motivation

When using TriggerDagRunOperator in synchronous mode (i.e., wait_for_completion=True), if the triggered child DAG fails (due to a failed task), the parent task also fails. However, when the operator is cleared in the UI (e.g., to retry), the associated failed child DAG run remains unchanged.

As a result:

• The retried TriggerDagRunOperator reuses the same failed DAG run,
• The child DAG run is still in failed state → the operator fails again,
• The user must manually clear failed tasks or the entire child DAG run.

This creates unnecessary friction when using DAG chaining.

Proposed solution:

Introduce a new optional parameter:

TriggerDagRunOperator(
    ...,
    auto_clear_failed_tasks=True
)

When set:

• Upon execution (or possibly in the clear() method logic),
• If a matching child DAG run exists and is in failed state,
• All failed task instances (and optionally downstreams) in that DAG run are automatically cleared.

This enables retrying the parent task to also retry the corresponding child DAG.

Benefits

• Smoother reprocessing of chained DAGs
• Reduces manual intervention
• Fits naturally with Airflow's existing retry model

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

[eladkal]

However, when the operator is cleared in the UI (e.g., to retry), the associated failed child DAG run remains unchanged.

This is not something that should be controlled by parameter in operator.
You should clear with recursive.

Recursive - All the tasks in the child dags and parent dags
https://airflow.apache.org/docs/apache-airflow/3.0.3/core-concepts/dag-run.html#re-run-tasks

@bbovenzi @pierrejeambrun I see that this option is not available in Airflow 3 but it does show up in the docs?
I see in #52130 it was mentioned as something we used to have as default for Airflow 2

[Pad71]

However, when the operator is cleared in the UI (e.g., to retry), the associated failed child DAG run remains unchanged.

This is not something that should be controlled by parameter in operator. You should clear with recursive.

Recursive - All the tasks in the child dags and parent dags https://airflow.apache.org/docs/apache-airflow/3.0.3/core-concepts/dag-run.html#re-run-tasks

@bbovenzi @pierrejeambrun I see that this option is not available in Airflow 3 but it does show up in the docs? I see in #52130 it was mentioned as something we used to have as default for Airflow 2

@eladkal
I believe that sometimes a user may simply want to reuse an existing child DAG run by clearing it entirely or prevent to clear child dag run at all rather than always triggering a new one. Therefore, instead of a single boolean parameter to control this behavior, it might make more sense to introduce a clear_mode option — allowing users to define how the TriggerDagRunOperator should behave when it's cleared.

This could support modes like:
"none" (default) – as-is behaviour
"failed_tasks" – clear only failed tasks in the triggered DAG
This would give users more flexibility when designing DAG chaining or orchestration patterns.

[eladkal]

The clear action may change depends on use case. It needs to be controlled from the UI not from operator code.
What am I missing?

[Pad71]

The clear action may change depends on use case. It needs to be controlled from the UI not from operator code. What am I missing?

@eladkal
The main point of this request is that currently, clearing a TriggerDagRunOperator task does not allow you to automatically clear failed tasks within the corresponding child DAG run.
This means that if one or more tasks in the child DAG fail, you must first manually fix or clear those failed tasks, then wait for the entire child DAG run to complete successfully, and only afterwards can you mark the TriggerDagRunOperator task as successful.

This creates a significant operational burden for the Operations team — especially in cases where the child DAG run is long-running.

Btw, even now we have triggerDagRunOperator parameter, which controls how the clear of TriggerDagRunOperator task is processed:

reset_dag_run (bool) – Whether or not clear existing dag run if already exists. This is useful when backfill or rerun an existing dag run. When reset_dag_run=False and dag run exists, DagRunAlreadyExists will be raised. When reset_dag_run=True and dag run exists, existing dag run will be cleared to rerun.

[pierrejeambrun]

I think Elad is right, and the closest we had in AF2 was recursive option in clear. I also agree that this shouldn't be controlled by the operator but by the UI.

Recursive controlled include_subdags and include_parentdag. (is_subdag, parent_dag, etc.) All that was removed with airflow 3 removal of subdags.

Also I don't think this would handle 'external triggers', only subdags (here i'm not sure), so this request might be slightly different and still make sense for AF3. (we want to clear external dependencies basically)

[Pad71]

I think Elad is right, and the closest we had in AF2 was recursive option in clear. I also agree that this shouldn't be controlled by the operator but by the UI.

Recursive controlled include_subdags and include_parentdag. (is_subdag, parent_dag, etc.) All that was removed with airflow 3 removal of subdags.

Also I don't think this would handle 'external triggers', only subdags (here i'm not sure), so this request might be slightly different and still make sense for AF3. (we want to clear external dependencies basically)

hello @pierrejeambrun Again, if it is done by UI, it's ok.
Problem is, that we can't do it now and it is realy painful for operations team (in our company we have lot's of dags triggered by triggerdagrunoperator tasks).
Recursive option does not work for that use case (in AF2 - 2.11. at the moment).

[Pad71]

I think, there are several use cases for TriggerDagRunOperator task Clear action:

Use Case 1: Do not clear child DAG run – raise exception

Description: The triggered DAG run is not cleared.

Behavior: If the DAG run is still in a failed or running state, the TriggerDagRunOperator task will raise an exception during the next attempt.

Current status: ✅ This is the current behavior when using reset_dag_run = False.

Purpose: Maintains strict separation and manual control between parent and child DAGs.

Use Case 2: Clear entire child DAG run

Description: When clearing the TriggerDagRunOperator, the entire child DAG run (triggered run) is also cleared.

Behavior: All tasks in the child DAG run are cleared (including successful ones).

The DAG run is reset and will be re-executed from scratch.
This may be time-consuming for large DAGs or long-running pipelines.

Current status: ✅ This is the default behavior when using reset_dag_run = True.

Purpose: Ensures full re-processing of the child DAG, regardless of current state.

Use Case 3: Clear only failed tasks in child DAG run (with downstream)

Description: When the TriggerDagRunOperator task is cleared, the system attempts to clear only failed tasks and their downstreams within the triggered child DAG run — without resetting the entire run.

Behavior (if implemented):
The child DAG run is not deleted, but failed tasks and their downstreams are cleared and re-executed.
This can reduce reprocessing time and resource usage significantly.

Current status: ❌ Not supported by Airflow out-of-the-box.

Workaround today:
The user must manually clear failed tasks in the child DAG (via UI/API/CLI).
Wait until the child DAG finishes successfully.

Then manually mark the TriggerDagRunOperator as successful.

Operations pain point:
If the child DAG run takes hours, this introduces significant overhead for Ops teams, who must monitor and intervene manually.

[pierrejeambrun]

Yes I think that could be a good feature to add, removing the needs-triage tag.

[pierrejeambrun]

@Pad71 Are you willing to work on that and open a PR for this ?

[Pad71]

@Pad71 Are you willing to work on that and open a PR for this ?

@pierrejeambrun I'm just a user, I'm not that familiar with Python to be able to do this :(