Versioned dag run uses the latest version of serialized dag in the Scheduler instead of sticking to their version #49007
Description
Apache Airflow version
3.0.0
If "Other Airflow 2 version" selected, which one?
No response
What happened?
Previously, when a DAG had multiple versions due to changes (e.g., a task removed), running DAG runs would sometimes incorrectly use the latest serialized DAG, even if they were started with an older version. This caused tasks that should still run to be incorrectly marked as removed.
The issue was caused by the scheduler not considering bundle_version when retrieving the DAG from the dagbag, defaulting to the latest available version.
What you think should happen instead?
No response
How to reproduce
Using the dag below in a git bundle, create a run, then comment out the first section, uncomment the second quickly while the first is still in the sleep task. Run the task and you will see the astronomer task marked removed.
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator
with DAG(dag_id="demo"):
# First run
sleep = BashOperator(task_id="sleep", bash_command="sleep 300")
hello = BashOperator(task_id="hello", bash_command="echo 'Hello'")
astronomer = BashOperator(task_id="astronomer", bash_command="echo 'Astonomer'")
sleep >> hello >> astronomer
# Second run
#sleep = BashOperator(task_id="sleep", bash_comma
Operating System
Linux
Versions of Apache Airflow Providers
No response
Deployment
Official Apache Airflow 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
- I agree to follow this project's Code of Conduct