apache / apache/airflow

ExternalTaskSensor waits forever if TaskGroup contains skipped tasks

Open
#33,053 8 comments 0 reactions 0 assignees View on GitHub
area:core area:TaskGroup kind:feature
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.