apache / apache/airflow

data-aware scheduling makes wrong dataset updates and downstream dag runs mapping when there are multiple updates during the execution of downstream dag

Open
#49,805 4 comments 0 reactions 0 assignees View on GitHub
area:core kind:bug needs-triage priority:medium
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 7h
Merged PRs (30d)
484

Description

### Apache Airflow version

2.10.5

### If "Other Airflow 2 version" selected, which one?

_No response_

### What happened?

I created a demo procedure where the upstream dag updates the dataset continuously, while the downstream dag just sleep 30s to simulate some time-consume tasks.

however, the dataset update and downstream dag run mappings are wired. it looks like below and there would be dangling dataset updates if I stop the upstream dag.

the dangling dataset updates are actually processed by downstream dag but it showed in ui that it would never be processed.

Image

Image

![Image](https://github.com/user-attachments/assets/154f94ae-6ba1-4de2-8e9d-b1466346a9a0)

### What you think should happen instead?

Image

### How to reproduce

```
from datetime import datetime as datetime
import airflow

from airflow import DAG

from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

from airflow.operators.python import PythonOperator
from airflow.operators.bash_operator import BashOperator

from airflow.datasets import Dataset

dataset1 = Dataset('s3://folder1/dataset_2.txt')

with DAG(
dag_id="upstream_dag_A",
start_date=datetime(2023, 1, 1),
catchup=False,
schedule="@continuous",
max_active_runs=1,
) as dag:
start_task = BashOperator(
task_id="start_task",
bash_command="echo 'Start task'",
outlets=[dataset1],
)

with DAG(
"downstream-dataset-dag",
start_date=datetime(2023, 1, 1),
schedule=[dataset1],
catchup=False,
max_active_runs=1,
) as dag:
start_task = BashOperator(
task_id="start_task",
# bash_command="echo 'Start task'",
bash_command='sleep 30 && echo "Upstream message: $message"',
)
```

### Operating System

ubuntu

### Versions of Apache Airflow Providers

_No response_

### Deployment

Other 3rd-party Helm chart

### 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

Open the contributing guide

Research direction

Start with the supplied two-DAG reproduction: an @continuous upstream DAG repeatedly updates one Dataset while the downstream DAG sleeps for 30 seconds. Compare the dataset-update and downstream-DAG-run mappings in the UI during multiple updates and after stopping the upstream DAG; done means processed updates are mapped correctly without dangling updates shown as unprocessed.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 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.