This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f6761c5cb80 Update waiting tasks to the latest Dag version (#74592)
f6761c5cb80 is described below
commit f6761c5cb80f4cd9a344f9ec07bab3237467e2c4
Author: Hussein Awala <[email protected]>
AuthorDate: Mon Oct 12 22:02:23 2026 +0200
Update waiting tasks to the latest Dag version (#74592)
Waiting task instances in unversioned Dag runs can retain stale version
references after a Dag update, even when their upstream tasks already reference
the new version. This leaves misleading version metadata when the downstream
tasks are eventually scheduled.
* Verify finished tasks retain their original Dag version
The regression test should reject an overly broad version update while
still proving that waiting tasks receive the latest version.
---
.../src/airflow/jobs/scheduler_job_runner.py | 2 +-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 49 ++++++++++++++++++++++
2 files changed, 50 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index c78bbf1d4bb..40e1fa976ff 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -3382,7 +3382,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.where(
TI.dag_id == dag_run.dag_id,
TI.run_id == dag_run.run_id,
- TI.state.in_(State.unfinished),
+ or_(TI.state.in_(State.unfinished), TI.state.is_(None)),
)
.values(dag_version_id=latest_dag_version.id),
execution_options={"synchronize_session": False},
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 530a5f99033..7118d08edc6 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -5957,6 +5957,55 @@ class TestSchedulerJob:
session.rollback()
session.close()
+ def test_verify_integrity_updates_dag_version_for_waiting_tasks(self,
dag_maker, session):
+ with dag_maker("test_version_update_with_waiting_task",
serialized=False, session=session) as dag:
+ upstream = BashOperator(task_id="upstream", bash_command="echo
upstream")
+ downstream = BashOperator(task_id="downstream", bash_command="echo
downstream")
+ BashOperator(task_id="finished", bash_command="echo finished")
+ upstream >> downstream
+
+ dr = dag_maker.create_dagrun()
+ assert dr.bundle_version is None
+ upstream_ti = dr.get_task_instance("upstream", session=session)
+ upstream_ti.state = State.RUNNING
+ downstream_ti = dr.get_task_instance("downstream", session=session)
+ assert downstream_ti.state is None
+ original_version_id = downstream_ti.dag_version_id
+ finished_ti = dr.get_task_instance("finished", session=session)
+ finished_ti.state = State.SUCCESS
+ assert finished_ti.dag_version_id == original_version_id
+ session.commit()
+
+ downstream.bash_command = "echo updated downstream"
+ SerializedDagModel.write_dag(
+ LazyDeserializedDAG.from_dag(dag),
bundle_name=dag_maker.bundle_name, session=session
+ )
+ session.commit()
+ latest_version_id = DagVersion.get_latest_version(dr.dag_id,
session=session).id
+ assert latest_version_id != original_version_id
+
+ self.job_runner = SchedulerJobRunner(job=Job())
+ self.job_runner._schedule_dag_run(dr, session)
+ session.commit()
+
+ session.refresh(upstream_ti)
+ session.refresh(downstream_ti)
+ session.refresh(finished_ti)
+ assert upstream_ti.dag_version_id == latest_version_id
+ assert downstream_ti.state is None
+ assert downstream_ti.dag_version_id == latest_version_id
+ assert finished_ti.state == State.SUCCESS
+ assert finished_ti.dag_version_id == original_version_id
+
+ upstream_ti.state = State.SUCCESS
+ session.commit()
+ self.job_runner._schedule_dag_run(dr, session)
+ session.commit()
+
+ session.refresh(downstream_ti)
+ assert downstream_ti.state == State.SCHEDULED
+ assert downstream_ti.dag_version_id == latest_version_id
+
def test_verify_integrity_not_called_for_versioned_bundles(self,
dag_maker, session):
with dag_maker("test_verify_integrity_if_dag_not_changed") as dag:
BashOperator(task_id="dummy", bash_command="echo hi")
