From 190e6c11747b293cc3cea73d3c4cf8659938058f Mon Sep 17 00:00:00 2001 From: Hanuman Singh Date: Tue, 14 Jul 2026 11:44:46 +0530 Subject: [PATCH] Handle TaskNotFound in triggerer to prevent stale triggers from blocking batch --- .../src/airflow/jobs/triggerer_job_runner.py | 15 ++++++-- .../tests/unit/jobs/test_triggerer_job.py | 36 +++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 355763aed5f13..6ea663d86dc22 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.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 @@ -879,12 +880,22 @@ def _create_workload( ) if serialized_dag_model: - task = serialized_dag_model.dag.get_task(trigger.task_instance.task_id) + try: + task = serialized_dag_model.dag.get_task(trigger.task_instance.task_id) + except TaskNotFound: + log.warning( + "Serialized Dag no longer contains the deferred task; " + "treating trigger as a context-free deferral", + trigger_id=trigger.id, + task_id=trigger.task_instance.task_id, + dag_version_id=trigger.task_instance.dag_version_id, + ) + task = None # 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) diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index 0a3354c122688..17808fd2bcd88 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -658,6 +658,42 @@ def test_create_workload_watched_assets_none_when_no_assets(jobless_supervisor, assert workload.watched_assets is None +def test_create_workload_handles_stale_task(jobless_supervisor, mocker): + """_create_workload() should return a context-free workload when get_task() raises TaskNotFound.""" + from airflow.exceptions import TaskNotFound + + trigger = mocker.Mock() + trigger.id = 99 + trigger.classpath = "some.path.Trigger" + trigger.encrypted_kwargs = "" + trigger.task_instance.dag_version_id = uuid.uuid4() + trigger.task_instance.task_id = "vanished_task" + trigger.task_instance.trigger_timeout = None + + mocker.patch( + "airflow.jobs.triggerer_job_runner.TaskInstanceDTO.model_validate", + return_value=mocker.Mock(spec=TaskInstanceDTO), + ) + + dag_bag = mocker.Mock() + serialized_dag_model = mocker.Mock() + serialized_dag_model.dag.get_task.side_effect = TaskNotFound("Task vanished_task not found") + dag_bag.get_serialized_dag_model.return_value = serialized_dag_model + + render_log_fname = mocker.Mock(return_value="/logs/ti") + + workload = jobless_supervisor._create_workload( + trigger=trigger, + dag_bag=dag_bag, + render_log_fname=render_log_fname, + session=mocker.Mock(), + ) + + assert workload is not None + assert workload.id == 99 + assert workload.dag_data is None + + def test_run_trigger_workload_includes_watched_assets_field(): """RunTrigger workload should accept and store watched_assets.""" workload = RunTrigger(