ExternalTaskSensor waits forever if TaskGroup contains skipped tasks
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Apache Airflow version
Other Airflow 2 version (please specify below)
### What happened
If you have an `ExternalTaskSensor` that uses `external_task_group_id` to wait on a `TaskGroup`, and if that `TaskGroup` contains any skipped tasks, the sensor will be stuck waiting forever despite the UI saying the state of the `TaskGroup` is successful.
### What you think should happen instead
`ExternalTaskSensor` should match the UI's interpretation of the `TaskGroup` state.
### How to reproduce
```
#!/usr/bin/env python3
import datetime
import logging
from airflow.decorators import dag, task
from airflow.operators.empty import EmptyOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup
from airflow.exceptions import AirflowFailException, AirflowSkipException
logger = logging.getLogger(__name__)
@dag(
schedule_interval='@daily',
start_date=datetime.datetime(2023, 8, 1),
)
def task_groups():
with TaskGroup(group_id='skip_group'):
@task
def skip_task():
raise AirflowSkipException
EmptyOperator(task_id='operator1') >> skip_task()
with TaskGroup(group_id='pass_group'):
@task
def pass_task():
pass
EmptyOperator(task_id='operator3') >> pass_task()
ExternalTaskSensor(
task_id='wait_for_task_group_with_skipped_task',
external_dag_id='task_groups',
external_task_group_id='skip_group',
check_existence=True,
)
ExternalTaskSensor(
task_id='wait_for_task_group_with_passed_task',
external_dag_id='task_groups',
external_task_group_id='pass_group',
check_existence=True,
)
dag = task_groups()
if __name__ == '__main__':
dag.cli()
```
### Operating System
CentOS Stream 8
### Versions of Apache Airflow Providers
_No response_
### Deployment
Other
### Deployment details
Standalone
### 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 with the ExternalTaskSensor implementation in airflow.sensors.external_task and trace how external_task_group_id resolves a TaskGroup's state when skipped tasks are present. Reproduce the DAG shown in the issue, then inspect the existing sensor tests if available and add coverage showing that a UI-successful group no longer waits forever.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100