Trigger Rule NONE_FAILED_MIN_ONE_SUCCESS and ONE_DONE not skipping a task if a Branch succeeds before
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow version
3.0.6
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
I have a basic dag where a task that should be skipped, isn't.
Here its whats happening:
- Essentially if `branch1` goes to `branch2` its logs seems correct it says `Following branch {'branch2'}` and `Skipping tasks []` which seems okay as we dont want to skip `problematic` as it might get executed via another dependency later
- `branch2` goes to `deadend` thus skipping dummy.
- **Even though `dummy` is skipped, `problematic` still occurs, even though it should not**
My guess is, even though `branch1` went to `branch2` it still technically `succeeded` as such it triggers `problematic` which has a `NONE_FAILED_MIN_ONE_SUCCESS` Trigger Rule
Dag code:
See in section below
A quick workaround is to put a `buffer` task so the that the we see that `buffer` actually is present on the Skipped tasks of `branch1` (see logs on the right). Although I don't this should really be needed.
### What you think should happen instead?
`problematic_task` with `TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS` should not execute, as the only upstream successfull task is a `task.branch` that went to another route
### How to reproduce
import the following dag. Verify that even when `deadend` is reached, the `problematic` task is still executing
```python
import datetime
import logging
import random
from airflow.decorators import dag, task
from airflow.utils.trigger_rule import TriggerRule
logger = logging.getLogger(__name__)
@dag(
start_date=datetime.datetime(2024, 5, 5),
schedule=None,
catchup=False,
)
def hep_create_dag():
@task.branch
def branch1(**context):
value = random.choice([True, False])
if value:
return "problematic"
return "branch2"
@task
def dummy(**context,
):
print("continue")
@task
def deadend(**context):
print("deadend")
@task.branch
def branch2(**context):
if False:
return "dummy"
return "deadend"
@task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
def problematic(**context):
print("problematic")
branch2_task = branch2()
problematic_task = problematic()
dummy_task = dummy()
deadend_task = deadend()
(
branch1()
>> [problematic_task, branch2_task]
)
branch2_task >> [ deadend_task,dummy_task,
]
(
dummy_task >> problematic_task
)
hep_create_dag()
```
### Operating System
docker in amd64 linux
### Versions of Apache Airflow Providers
default 3.0.6 installation
### Deployment
Docker-Compose
### 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
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start by running the supplied DAG on Airflow 3.0.6 and inspect the trigger-rule and branch behavior involving branch1, branch2, dummy, deadend, and problematic. Done means problematic_task does not execute when the only successful upstream route is branch1's alternate branch and deadend is reached.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 45/100