flinz commented on issue #30883:
URL: https://github.com/apache/airflow/issues/30883#issuecomment-1604262465
Can confirm this still exists in 2.6.0 and I independently hit this using:
```python
import time
from datetime import datetime
from airflow import DAG
from airflow.decorators import task, task_group
from airflow.operators.empty import EmptyOperator
from airflow.utils.trigger_rule import TriggerRule
branches = [1, 2, 3]
with DAG(
dag_id="test-map-branching",
schedule=None,
catchup=False,
start_date=datetime(2023, 1, 1),
default_args={"retries": 0},
) as dag:
@task_group(group_id="branch_run")
def mapped_group(branch_number):
"""Process single sample"""
@task.branch(dag=dag)
def choose_branch(branch_number, **kwargs):
print(branch_number)
if branch_number == 1:
return "branch_run.first"
else:
return "branch_run.other"
choose_tech_branch_result = choose_branch(branch_number)
@task(dag=dag)
def first(branch_number, **kwargs):
print(branch_number)
first_result = first(branch_number)
choose_tech_branch_result >> first_result
@task(dag=dag)
def other(branch_number, **kwargs):
print(branch_number)
if branch_number == 3:
time.sleep(5)
return branch_number
other_result = other(branch_number)
choose_tech_branch_result >> other_result
@task(dag=dag, trigger_rule=TriggerRule.NONE_FAILED)
def finalize_map(branch_number, **kwargs):
print(branch_number)
return branch_number
finalize_map_result = finalize_map(branch_number)
[first_result, other_result] >> finalize_map_result
return finalize_map_result
@task(dag=dag)
def finalize_all(group_results):
print(group_results)
group_results = mapped_group.expand(branch_number=branches)
finalizer_result = finalize_all(group_results)
```
This always skips downstream branch tasks
--
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]
