Premature task triggering when using dynamic task mapping and task groups

[BTurkowyd]

When using dynamic task mapping and task groups, we noticed that the tasks in the task group, when their trigger rule is set to ONE_SUCCESS or NONE_FAILED_MIN_ONE_SUCCESS, are being triggered (and skipped) immediately at the start of the task group execution. As we are doing some branching and joining within the task group, we would like to use on of these trigger rules, but currently tasks will be skipped prematurely.

On the screenshot below I present the graph I prepared to test the issue. task_10 and task_12 have NONE_FAILED_MIN_ONE_SUCCESS as they are joining tasks waiting for the execution of upstream tasks. As you can see, they are immediately skipped, even when the task_3 which is the first task within the task group is being queued for execution. I also checked that task_8 and task_9 were not executed in any of the mapped instances.

Here are the start times of:
Task_10: 2024-05-27, 12:43:44 CEST
Task_12: 2024-05-27, 12:43:44 CEST
Task_3: 2024-05-27, 12:43:51 CEST (7 seconds after tasks 10 and 12)

I found a similar issue #28973 and tried a solution proposed there, but it didn’t help in our case.
Finally, I tested this dagger locally on Airflow 2.8.2 and the outcome is the same.

Below is the code for the DAG from the screenshot.

My question is: Am I doing something wrong, do I miss something in the code to prevent such a behavior? At the moment we agreed we will use ALL_DONE trigger, but we would prefer to stick to NONE_FAILED_MIN_ONE_SUCCESS.

Thank you in advance for your help!

from datetime import datetime

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


def get_xcom_value(kwargs, task_id: str, key_id: str = None):
    return kwargs['ti'].xcom_pull(task_ids=task_id, key=key_id)

@dag(
    tags=['test'],
    catchup=False,
    start_date=datetime(2024, 4, 12),
    default_args={
        "retries": 0,
    })
def minimal_dag_task_mapping():
    @task
    def task_1(**context):
        print('I am the first task')

    @task
    def task_2(**context):
        to_map = [i for i in range(4)]
        print('I am the second task')
        return to_map

    @task_group()
    def tg(mapping):
        @task.branch()
        def task_3(**context):
            mapping_list = mapping.resolve(context)
            print(mapping_list)
            force_to_branch = True

            if force_to_branch:
                return 'tg.task_5'

            return 'tg.task_6'

        @task.branch()
        def task_5(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')
            force_to_branch = True

            if force_to_branch:
                return 'tg.task_8'

            return 'tg.task_9'

        @task
        def task_6(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        @task
        def task_8(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        @task
        def task_9(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
        def task_10(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        @task
        def task_11(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
        def task_12(**context):
            map_index = context['ti'].map_index
            print(f'Map index: {map_index}.')

        task_3 = task_3()
        task_5 = task_5()
        task_6 = task_6()
        task_8 = task_8()
        task_9 = task_9()
        task_10 = task_10()
        task_11 = task_11()
        task_12 = task_12()
        task_3 >> [task_5, task_6]
        task_5 >> [task_8, task_9]
        task_8 >> task_10
        task_9 >> task_10
        task_10 >> task_11 >> task_12
        task_6 >> task_12

    task_1 = task_1()
    task_2 = task_2()
    tg = tg.expand(mapping=task_2)

    task_1 >> task_2 >> tg

minimal_dag_task_mapping()

[raphaelauv]

#39801