Unexpected behavior of dynamically generated task with TriggerRule ONE_FAILED

[jmelot]

Apache Airflow version

2.10.5

If "Other Airflow 2 version" selected, which one?

No response

What happened?

We have a dynamically generated task inside a task group with trigger rule ONE_FAILED and an upstream task that fails in some cases and succeeds in others. We find that if one instance of the upstream task fails and others succeed, the task task with trigger rule ONE_FAILED is marked success in all instances rather than being skipped in some and marked success in others.

What you think should happen instead?

I would expect this task to be marked skipped when no upstream task fails.

How to reproduce

I made this simple dag to demonstrate the behavior we are seeing

import logging

from airflow.decorators import task, task_group
from airflow.decorators import dag
from airflow.utils.trigger_rule import TriggerRule


logger = logging.getLogger(__name__)

@task
def divide(i):
    return 30/i

@task(trigger_rule=TriggerRule.ONE_FAILED)
def report_failure(i):
    logger.info(f"{i} failed")

@task
def report_success(i):
    logger.info(f"{i} succeeded")

@dag(
    dag_id="Test-Trigger-Rule",
    schedule=None,
    catchup=False,
)
def test_trigger_rule_dag():
    @task
    def gen_examples():
        return [0,1,2,3]
    
    @task_group
    def divide_and_report(i):
        divide(i) >> [report_success(i), report_failure(i)]
        
    divide_and_report.expand(i=gen_examples())

test_trigger_rule_dag()

When I run this dag, report_success behaves as I would expect, with three mapped tasks in success and one in upstream_failed. report_failure is always marked success, however, rather than three tasks being marked skipped and one success.

If this is a bug I'm happy to do what I can to help fix it. If I've misunderstood something and this behavior is actually expected, I'm happy to help update documentation if needed so others aren't confused.

Operating System

macOS 15.4.1

Versions of Apache Airflow Providers

No response

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[eladkal]

@shahar1 You've done some work around this. WDYT?

[shahar1]

@shahar1 You've done some work around this. WDYT?

It seems indeed like a bug -

The ONE_FAILED trigger rule evaluates the state of the entire upstream task, and not its specific mapped tasks, see following:

airflow/airflow/ti_deps/deps/trigger_rule_dep.py

Line 410 in a0062c8

elif trigger_rule == TR.ONE_FAILED:

@jmelot you're welcome to explore it and come up with a PR.
I'll try to give it some thought myself, and keep you posted if I have an idea :)

[shahar1]

I went over the code again in Airflow v2 (v3 is quite the same), and it seems that changing the existing behavior is not be as trivial - it might have been even an intended behavior, at least at the time mapped tasks were originally introduced.
The current design prioritizes lazy evaluation of mapped tasks rather than querying the state of their mapped upstream(s), and it contains some heavy logic to support it (get_relevant_upstream_map_indexes). In my recent PR in this area, I had introduced a fix for premature evaluation of the "fast triggered" trigger rules (see #44937, once_failed included) - which technically makes all TIs of a specific upstream dependencies (otherwise, upcoming task is auto-skipped). Adjusting it to check the state of each upstream with the appropriate map_index might be more complicated.

Some intermediate conclusions:

1. We should think of a better mechanism to combine trigger rules and mapped tasks (and refactor trigger_rule_dep.py accordingly).
2. Following the above, the terminology "ONE_"/"NONE_" is confusing in the context mapped tasks in the current implementation. Renaming them to comply with a more flexible behavior should be done as part of a larger refactoring effort.

@uranusjr, @dstandish - if you have any thoughts or ideas regarding this, please feel free to share.

[uranusjr]

It’s not exactly intentionally designed, but more like we didn’t know what the correct design should be, so let’s just roll with it and figure things out. I guess we’re about at the point of figuring it out now.

I don’t really have a set idea around this; I can think of valid use cases for either treating all expanded tis of a mapped task as one entity, or each ti as a separate entity. It’s however likely that we can figure out one set of rules that is more useful to most people.