joined task always skipped

[olk]

Apache Airflow version

2.10.4

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

No response

What happened?

The joning task after branching is always skipped with trigger policy="none_failed_min_one_success":

from airflow.utils.dates import days_ago
from airflow.decorators import dag, task, task_group

@task
def start():
    return [ 1, 2, 3 ]


@task_group
def expanding(n):

    @task.branch
    def branching(n):
        print(n)

        if n == 1:
            return "expanding.foo"
        else:
            return "expanding.bar"

    @task
    def foo(n):
        print(n)
        return n + 1

    @task
    def bar(n):
        print(n)
        return n + 3

    @task(trigger_rule='none_failed_min_one_success')
    def finalize(x, y):
        if x is not None:
            return x
        if y is not None:
            return y
        return None

    choose_tech_branch_result = branching(n)

    foo_result = foo(n)
    bar_result = bar(n)

    choose_tech_branch_result >> [ foo_result, bar_result ]

    finalize_result = finalize(foo_result, bar_result)
    [ foo_result, bar_result ] >> finalize_result

    return finalize_result


@task
def finalize_all(group_results):
    print(group_results)
    return group_results


@dag(dag_id="branching",
     start_date=days_ago(1),
     schedule_interval=None)
def branch_example():

    group_results = expanding.expand(n=start())
    finalizer_result = finalize_all(group_results)
    print(finalizer_result)

branch_example()

Workaround: pass argument n to task ´finalize` even if it's not used. seams to be an issue with upstream evaluation.

from airflow.utils.dates import days_ago
from airflow.decorators import dag, task, task_group

@task
def start():
    return [ 1, 2, 3 ]


@task_group
def expanding(n):

    @task.branch
    def branching(n):
        print(n)

        if n == 1:
            return "expanding.foo"
        else:
            return "expanding.bar"

    @task
    def foo(n):
        print(n)
        return n + 1

    @task
    def bar(n):
        print(n)
        return n + 3

    @task(trigger_rule='none_failed_min_one_success')
    def finalize(_, x, y):
        if x is not None:
            return x
        if y is not None:
            return y
        return None

    choose_tech_branch_result = branching(n)

    foo_result = foo(n)
    bar_result = bar(n)

    choose_tech_branch_result >> [ foo_result, bar_result ]

    finalize_result = finalize(n, foo_result, bar_result)
    [ foo_result, bar_result ] >> finalize_result

    return finalize_result


@task
def finalize_all(group_results):
    print(group_results)
    return group_results


@dag(dag_id="branching",
     start_date=days_ago(1),
     schedule_interval=None)
def branch_example():

    group_results = expanding.expand(n=start())
    finalizer_result = finalize_all(group_results)
    print(finalizer_result)

branch_example()

What you think should happen instead?

No response

How to reproduce

see comment/code

Operating System

Linux

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

[potiuk]

Please use discusion for such question - possibly someone can help you, but it's not really clear if it is Airflow issue.

[olk]

Why is it a 'discussion'? I think it is a bug - trigger_rule 'none_failed_min_one_success' always skips the joining downstream task.

The expected behavior is that the joining task is always executed and not skipped - otherwise 'none_failed_min_one_success' would not make sense!

[olk]

Please use discusion for such question - possibly someone can help you, but it's not really clear if it is Airflow issue.

Bug #39801 seams to be related - so please move it back to the issue list.

[potiuk]

I think you are missing the point of issue vs. discussion/

Discussion is not "worse" than issue. In order to get it implemented someone has to have clarity if this is Airflow issue or not. And hopefully - discussion here will lead to that outcome. And if you see there is a related issue - you should comment there - point to that discussion, and hopefully help to reach the conclusion about the issue and it's scope.

Then, somene will have to anyhow decide on their own to work on that issue. On their own. As volunteer. The fact whether it's an isue or discussion does not matter, and does not change the chance of things being worked on. This is volunteer, open-source driven project, so you should not treat issues as "list of things that needs to be done". This is basically something maintainers had enough information on to classify as issue. That's it.

The fastest way to fix some problem is to contribute a PR fixing is. You are most welcome. Second best is to convince someone it's worth their time and pick their interest.

Both of these can be done in issue or discussion and discussion can be converted to issue if there is enough understanding and clerity it is indeed something problematic in Airflow.