This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new a3b282bc8dd [v3-3-test] Load the correct DAG version when a task
starts from a trigger (#69988) (#72614)
a3b282bc8dd is described below
commit a3b282bc8ddac6792da859aefac8f7c86e5c3691
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Sep 8 14:40:32 2026 +0530
[v3-3-test] Load the correct DAG version when a task starts from a trigger
(#69988) (#72614)
* Pin triggerer start_from_trigger DAG resolution to the dagrun's version
* Resolve trigger Dag version via _version_from_dag_run for pinned/unpinned
parity
* Skip triggers whose task is missing from the resolved Dag version
For an unpinned run the resolved (latest) Dag version may no longer
contain a renamed or removed deferred task. get_task() then raised
TaskNotFound, which nothing on the load_triggers path caught, killing
the whole triggerer; assign_unassigned re-handed the same trigger to the
restarted triggerer, producing a crash loop. Guard the lookup and fall
through to the plain workload, mirroring the guard in the execution API.
Also introduce DBDagBag.get_serialized_dag_model_for_run so the
triggerer uses a public API instead of the private _version_from_dag_run
helper, as requested in review.
(cherry picked from commit aead44307292bb3a927504d5ebcd9e3104b624aa)
Co-authored-by: Sean Ghaeli <[email protected]>
---
.../src/airflow/jobs/triggerer_job_runner.py | 22 ++++--
airflow-core/src/airflow/models/dagbag.py | 8 ++
airflow-core/tests/unit/jobs/test_triggerer_job.py | 88 +++++++++++++++++++++-
3 files changed, 111 insertions(+), 7 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
index b8788c05e45..ac8bc52370c 100644
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
@@ -50,6 +50,7 @@ from airflow._shared.module_loading import import_string
from airflow._shared.observability.metrics import stats
from airflow._shared.timezones import timezone
from airflow.configuration import conf
+from airflow.exceptions import TaskNotFound
from airflow.executors import workloads
from airflow.executors.workloads.task import TaskInstanceDTO
from airflow.jobs.base_job_runner import BaseJobRunner
@@ -862,20 +863,29 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
ti=ser_ti, # type: ignore
)
- serialized_dag_model = dag_bag.get_serialized_dag_model(
- version_id=trigger.task_instance.dag_version_id,
- session=session,
+ dag_run = trigger.task_instance.get_dagrun(session=session)
+ serialized_dag_model = dag_bag.get_serialized_dag_model_for_run(
+ dag_run, session=session
+ ) or dag_bag.get_serialized_dag_model(
+ version_id=trigger.task_instance.dag_version_id, session=session
)
if serialized_dag_model:
- task =
serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
+ task = None
+ try:
+ task =
serialized_dag_model.dag.get_task(trigger.task_instance.task_id)
+ except TaskNotFound:
+ log.warning(
+ "Task not found in resolved Dag version; building plain
workload",
+ task_id=trigger.task_instance.task_id,
+ dag_id=trigger.task_instance.dag_id,
+ )
# When a TaskInstance of a Trigger contains a task with
start_from_trigger enabled,
# it means we need to load the SerializedDagModel so we can build
a RuntimeTaskInstance later on which
# will allow us to build a context on which we will render the
templated fields.
- if task.start_from_trigger:
+ if task is not None and task.start_from_trigger:
log.info("Start from trigger enabled for task %s",
task.task_id)
- dag_run = trigger.task_instance.get_dagrun(session=session)
return workloads.RunTrigger(
id=trigger.id,
diff --git a/airflow-core/src/airflow/models/dagbag.py
b/airflow-core/src/airflow/models/dagbag.py
index 92662d067a8..1665699bec1 100644
--- a/airflow-core/src/airflow/models/dagbag.py
+++ b/airflow-core/src/airflow/models/dagbag.py
@@ -208,6 +208,14 @@ class DBDagBag:
return self._get_dag(version_id=version_id, session=session)
return None
+ def get_serialized_dag_model_for_run(
+ self, dag_run: DagRun, *, session: Session
+ ) -> SerializedDagModel | None:
+ """Return the SerializedDagModel for the dag version this dag run
resolves to."""
+ if version_id := self._version_from_dag_run(dag_run=dag_run,
session=session):
+ return self.get_serialized_dag_model(version_id=version_id,
session=session)
+ return None
+
def iter_all_latest_version_dags(self, *, session: Session) ->
Generator[SerializedDAG, None, None]:
"""
Walk through all latest version dags available in the database.
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index c354f6cf17f..a4181b44ea2 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -68,6 +68,7 @@ from airflow.jobs.triggerer_job_runner import (
)
from airflow.models import Connection, DagModel, DagRun, Trigger, Variable
from airflow.models.dag_version import DagVersion
+from airflow.models.dagbag import DBDagBag
from airflow.models.dagbundle import DagBundleModel
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.xcom import XComModel
@@ -615,7 +616,7 @@ def
test_create_workload_uses_supervisor_id_without_job(jobless_supervisor, mock
serialized_dag_model = mocker.Mock()
task = mocker.Mock(start_from_trigger=False)
serialized_dag_model.dag.get_task.return_value = task
- dag_bag.get_serialized_dag_model.return_value = serialized_dag_model
+ dag_bag.get_serialized_dag_model_for_run.return_value =
serialized_dag_model
render_log_fname = mocker.Mock(return_value="/logs/ti")
@@ -630,6 +631,91 @@ def
test_create_workload_uses_supervisor_id_without_job(jobless_supervisor, mock
assert factory.log_path == f"/logs/ti.trigger.{jobless_supervisor.id}.log"
[email protected](
+ "pinned", [True, False], ids=["pinned-uses-run-created-version",
"unpinned-uses-latest-version"]
+)
+def test_create_workload_resolves_serialized_dag_from_run(jobless_supervisor,
mocker, pinned):
+ """The trigger should load the run's Dag version: created version if
pinned, latest otherwise."""
+ run_created_version = uuid.uuid4()
+ latest_version = uuid.uuid4()
+ bumped_ti_version = uuid.uuid4()
+
+ trigger = mocker.Mock()
+ trigger.id = 8
+ trigger.classpath = "some.path.Trigger"
+ trigger.encrypted_kwargs = ""
+ trigger.task_instance.dag_version_id = bumped_ti_version
+ trigger.task_instance.task_id = "t"
+ trigger.task_instance.trigger_timeout = None
+
+ dag_run = mocker.Mock(spec=DagRun)
+ dag_run.dag_id = "test_dag"
+ dag_run.bundle_version = "some-bundle-version" if pinned else None
+ dag_run.created_dag_version_id = run_created_version
+ dag_run.dag_run_data = mocker.Mock()
+ dag_run.dag_run_data.model_dump.return_value = {}
+ trigger.task_instance.get_dagrun.return_value = dag_run
+
+ mocker.patch.object(
+ DagVersion, "get_latest_version",
return_value=mocker.Mock(spec=DagVersion, id=latest_version)
+ )
+ mocker.patch(
+ "airflow.jobs.triggerer_job_runner.TaskInstanceDTO.model_validate",
+ return_value=mocker.Mock(spec=TaskInstanceDTO),
+ )
+
+ dag_bag = DBDagBag()
+ serialized_dag_model = mocker.Mock()
+ task = mocker.Mock(start_from_trigger=True)
+ serialized_dag_model.dag.get_task.return_value = task
+ serialized_dag_model.data = {}
+ mocker.patch.object(dag_bag, "get_serialized_dag_model",
return_value=serialized_dag_model)
+
+ session = mocker.Mock()
+ jobless_supervisor._create_workload(
+ trigger=trigger,
+ dag_bag=dag_bag,
+ render_log_fname=mocker.Mock(return_value="/logs/ti"),
+ session=session,
+ )
+
+ expected_version = run_created_version if pinned else latest_version
+
dag_bag.get_serialized_dag_model.assert_called_once_with(version_id=expected_version,
session=session)
+
+
+def
test_load_triggers_survives_task_missing_from_resolved_dag_version(supervisor_builder,
session, caplog):
+ """
+ A deferred TI's task may be missing from the Dag version an unpinned run
resolves to
+ (latest), e.g. after the task was renamed. TaskNotFound must not escape
workload
+ building — it previously killed the whole triggerer, and assign_unassigned
re-handing
+ the trigger to the restarted triggerer produced a crash loop. Instead the
trigger
+ gets a plain workload (no dag_data) and a warning is logged.
+ """
+ trigger = TimeDeltaTrigger(datetime.timedelta(days=7))
+ _, run, trigger_orm, _ = create_trigger_in_db(session, trigger)
+ assert run.bundle_version is None # unpinned run resolves to the latest
version
+
+ # The Dag is edited: the deferred task is renamed away in the new latest
version
+ dag_v2 = DAG(dag_id="test_dag", schedule="@daily",
start_date=pendulum.datetime(2023, 1, 1))
+ BaseOperator(task_id="renamed_ti", dag=dag_v2)
+ SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag_v2),
bundle_name="testing")
+ session.commit()
+
+ job = Job(heartrate=10)
+ job.job_type = "TriggererJob"
+ job.latest_heartbeat = timezone.utcnow()
+ session.add(job)
+ session.flush()
+ supervisor = supervisor_builder(job=job)
+ session.commit()
+
+ supervisor.load_triggers()
+
+ workload = next(w for w in supervisor.creating_triggers if w.id ==
trigger_orm.id)
+ assert workload.dag_data is None
+ assert "Task not found in resolved Dag version; building plain workload"
in caplog
+
+
def
test_create_workload_sets_watched_assets_for_asset_only_trigger(jobless_supervisor,
mocker):
"""_create_workload() should populate watched_assets when
trigger.task_instance is None and assets exist."""
asset1 = mocker.Mock(spec=Asset)
