joshOberhaus commented on issue #65745:
URL: https://github.com/apache/airflow/issues/65745#issuecomment-4338200248
@MFirulik, for an `airflow standalone` bring-up with Airflow 3.2.1, I am not
seeing the undesired 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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
