Branch task does not prevent non-selected downstream tasks from running in TaskGroup (Airflow 3.2.x)

[MFirulik]

Under which category would you file this issue?

Providers

Apache Airflow version

3.2.0 (also reproduced on 3.2.1)

What happened and how to reproduce it?

Branching inside a TaskGroup does not reliably prevent non-selected downstream tasks from running.
Even when branch callable returns a specific task_id, sibling downstream tasks still get scheduled/run.
In our production setup (KubernetesExecutor), this causes unnecessary pod starts and resource waste.

Minimal example (TaskFlow + TaskGroup + branch):

from airflow.sdk import DAG, task, task_group
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk.bases.operator import chain
from pendulum import datetime
with DAG(
    dag_id="repro_branch_taskgroup_32x",
    start_date=datetime(2026, 1, 1),
    schedule=None,
    catchup=False,
):
    @task_group(group_id="g")
    def g():
        @task.branch(task_id="choose")
        def choose():
            return "g.path_a"
        @task(task_id="path_a")
        def path_a():
            print("A")
        @task(task_id="path_b")
        def path_b():
            print("B should be skipped")
        end = EmptyOperator(task_id="end")
        c = choose()
        a = path_a()
        b = path_b()
        chain(c, [a, b], end)
    g()

What you think should happen instead?

Only selected downstream branch task(s) should run.
All non-selected downstream tasks should be marked skipped before execution.

Operating System

No response

Deployment

Official Apache Airflow Helm Chart

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

1.18.0

Kubernetes Version

v1.32.4

Helm Chart configuration

No response

Docker Image customizations

No response

Anything else?

Environment
Executor: KubernetesExecutor
Python: 3.13
Deployment: Docker/Kubernetes
DAG contains TaskGroup and mapped workloads in real case

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.

[MFirulik]

Hi. Yes, no problem. I'd be very happy! чт, 23 апр. 2026 г. в 22:02, Md.Sadiq ***@***.***>:

…

*MdSadiqMd* left a comment (apache/airflow#65745) <#65745 (comment)> Hey @MFirulik <https://github.com/MFirulik> are you working on it? If not can i take it — Reply to this email directly, view it on GitHub <#65745?email_source=notifications&email_token=BFZWDVRHP5QQOX7IBNAZPRL4XJZEZA5CNFSNUABFM5UWIORPF5TWS5BNNB2WEL2JONZXKZKDN5WW2ZLOOQXTIMZQG42TGNJXHE2KM4TFMFZW63VHNVSW45DJN5XKKZLWMVXHJLDGN5XXIZLSL5RWY2LDNM#issuecomment-4307535794>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/BFZWDVXEJ52RDEE7WEEK5NT4XJZEZAVCNFSM6AAAAACYEI2XKCVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHM2DGMBXGUZTKNZZGQ> . Triage notifications, keep track of coding agent tasks and review pull requests on the go with GitHub Mobile for iOS <https://github.com/notifications/mobile/ios/BFZWDVUKUCUS5Q2CAOQQRSD4XJZEZA5CNFSNUABFM5UWIORPF5TWS5BNNB2WEL2JONZXKZKDN5WW2ZLOOQXTIMZQG42TGNJXHE2KM4TFMFZW63VHNVSW45DJN5XKKZLWMVXHJKTGN5XXIZLSL5UW64Y> and Android <https://github.com/notifications/mobile/android/BFZWDVRANTDVJFZ6OJMEBMD4XJZEZA5CNFSNUABFM5UWIORPF5TWS5BNNB2WEL2JONZXKZKDN5WW2ZLOOQXTIMZQG42TGNJXHE2KM4TFMFZW63VHNVSW45DJN5XKKZLWMVXHJLTGN5XXIZLSL5QW4ZDSN5UWI>. Download it today! You are receiving this because you were mentioned.Message ID: ***@***.***>

[MdSadiqMd]

Yeah, working on it, thanks

[joshOberhaus]

@MFirulik, for an airflow standalone bring-up with Airflow 3.2.1, I am not seeing the unexpected behavior, path_b skips, as expected. Am I missing something?
However, I notice that you state that your real DAG also has mapped workload, with mapped tasks, I am seeing this which I can show in this example:

"""Minimal example: @task.branch inside a dynamically mapped task group.
This example replicates the structure of a real-world DAG, where:
- One outer task group (process_items) fetches items and expands
    an inner task group (handle_item) via expand_kwargs.
- Inside handle_item, a @task.branch decides whether to run
    run_optional_step based on item_type.

Expected behavior
------------------
item_type == "full"    → branch returns absolute task ID → run_optional_step should run
item_type == "partial" → branch returns None             → run_optional_step should be skipped

Actual behavior
-------------------------
run_optional_step runs for ALL items.
"""

from __future__ import annotations

import pendulum
from airflow.sdk import DAG, task, task_group
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG(
    dag_id="branch_in_mapped_taskgroup_example",
    schedule=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
) as dag:

    # ------------------------------------------------------------------ #
    # Inner task group: handles one mapped item.                           #
    # The branch should skip or run an optional step per mapped instance.  #
    # ------------------------------------------------------------------ #
    @task_group(group_id="handle_item")
    def tg_handle_item(item_id: str, item_type: str):
        """Process one item.  Only 'full' items need the extra step."""

        @task(task_id="process")
        def process(item_id: str, item_type: str) -> str:
            print(f"Processing {item_id=} {item_type=}")
            return item_type

        result = process(item_id, item_type)

        @task.branch(task_id="select_optional_step")
        def select_optional_step(item_type: str):
            if item_type == "full":
                return ["process_items.handle_item.run_optional_step"]
            return None

        branch = select_optional_step(result)
        run_optional_step = EmptyOperator(task_id="run_optional_step")
        branch >> run_optional_step

    # ------------------------------------------------------------------ #
    # Outer task group: loads items and expands the inner group once per  #
    # item. There is only one outer group instance in the DAG.            #
    # ------------------------------------------------------------------ #
    @task_group(group_id="process_items")
    def tg_process_items():
        @task(task_id="get_items")
        def get_items() -> list[dict]:
            return [
                {"item_id": "item_1", "item_type": "full"},     # optional step should run
                {"item_id": "item_2", "item_type": "partial"},  # optional step should be skipped
            ]

        tg_handle_item.expand_kwargs(get_items())

    tg_process_items()

This results in skipping in 3.1.7, in 3.2.1, optional step is ran for both. Does this match with the behavior you're reporting? If not, I'll open a new bug ticket. Thanks!

I believe the behavior I'm describing was introduced with #65745 #62287,

[bretlourd-emtelligent]

@joshOberhaus we are experiencing the same behaviour as you with branching in mapped task groups. Definitely worth opening a new bug if not already addressed here

[martijn-exads]

We're hitting the same behaviour on branch operators inside expand/expand_kwargs-mapped TaskGroups in 3.2.x — non-selected branch siblings run instead of being skipped, and #65751 does not address this case (it fixes relative branch IDs in non-mapped TaskGroups; our branches already return absolute IDs).

We think we've traced it to #62287, specifically this line in NotPreviouslySkippedDep:
xcom_map_index = ti.map_index if parent.is_mapped else -1

For a branch operator inside a mapped @task_group, the operator itself has no .expand() so parent.is_mapped is False, but its TIs run with map_index >= 0 and SkipMixin writes XCOM_SKIPMIXIN_KEY at the parent TI's per-map map_index. The dep then queries map_indexes=-1, finds nothing, returns "not skipped", and the non-selected sibling proceeds.

Let us know whether this should stay on this ticket or be split into a separate one, happy to open one if that's preferred.

[joshOberhaus]

@bretlourd-emtelligent , @martijn-exads , thanks for the confirmation. Per @bretlourd-emtelligent 's suggestion, I have opened a new issue: #67265 (which mostly uses @martijn-exads formulation of the problem, which is far better than mine).

Also, @martijn-exads , yes, that is the PR I had traced it too as well, but I guess I made a copy and paste error as I linked this issue instead.

As for this one: I think perhaps based on the description, it was the same issue, but unfortunately as the repro code does not demonstrate the issue, perhaps it would be better to close it.