The TI doesn't wait setup tasks when they are not a direct upstream #33561
Description
Apache Airflow version
2.7.0
What happened
When we add a setup task (let's call it S) before a sequence of tasks, and we clear the state of a not direct downstream task for the setup task (let's call it T), the state of S is correctly cleared, but the task T doesn't wait for S, they are running in parallel.
What you think should happen instead
The setup task should be treated as an upstream for all the downstream tasks
How to reproduce
from datetime import datetime
from airflow.models import DAG
from airflow.operators.python import PythonOperator
def sleep_and_log(msg):
import time
time.sleep(5)
print(msg)
with DAG(
dag_id="test_setup_teardown",
schedule_interval=None,
start_date=datetime(2023, 8, 20),
catchup=False,
):
setup_task = PythonOperator(
task_id="setup_task",
python_callable=lambda: sleep_and_log("setup_task"),
).as_setup()
teardown_task = PythonOperator(
task_id="teardown_task",
python_callable=lambda: sleep_and_log("teardown_task"),
).as_teardown(setups=[setup_task])
t1 = PythonOperator(
task_id="t1",
python_callable=lambda: sleep_and_log("t1"),
)
t2 = PythonOperator(
task_id="t2",
python_callable=lambda: sleep_and_log("t2"),
)
setup_task >> t1 >> t2 >> teardown_taskCreate a run and wait all the tasks to finish, then clear the state of t2, the setup task and t2 states will be cleared, and both tasks will run in parallel.
Operating System
Debian GNU/Linux
Versions of Apache Airflow Providers
No response
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
- I agree to follow this project's Code of Conduct