diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 8719f4ea95c8f..79b5e953dcd49 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -680,7 +680,14 @@ def update_triggers(self, requested_trigger_ids: set[int]): # Either the trigger code or the path to it is bad. Fail the trigger. self.failed_triggers.append((new_id, e)) continue - new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) + + try: + new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) + except TypeError as err: + self.log.error("Trigger failed; message=%s", err) + self.failed_triggers.append((new_id, err)) + continue + self.set_trigger_logging_metadata(new_trigger_orm.task_instance, new_id, new_trigger_instance) self.to_create.append((new_id, new_trigger_instance)) # Enqueue orphaned triggers for cancellation diff --git a/tests/jobs/test_triggerer_job.py b/tests/jobs/test_triggerer_job.py index 73b8d878a267d..fc08b8a5d0803 100644 --- a/tests/jobs/test_triggerer_job.py +++ b/tests/jobs/test_triggerer_job.py @@ -37,7 +37,7 @@ from airflow.operators.empty import EmptyOperator from airflow.operators.python import PythonOperator from airflow.triggers.base import TriggerEvent -from airflow.triggers.temporal import TimeDeltaTrigger +from airflow.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger from airflow.triggers.testing import FailureTrigger, SuccessTrigger from airflow.utils import timezone from airflow.utils.log.logging_mixin import RedirectStdHandler @@ -285,6 +285,26 @@ async def test_run_trigger_timeout(self, session, caplog) -> None: await trigger_runner.run_trigger(1, mock_trigger) assert "Trigger cancelled due to timeout" in caplog.text + @patch("airflow.models.trigger.Trigger.bulk_fetch") + @patch( + "airflow.jobs.triggerer_job_runner.TriggerRunner.get_trigger_by_classpath", + return_value=DateTimeTrigger, + ) + def test_update_trigger_with_triggerer_argument_change( + self, mock_bulk_fetch, mock_get_trigger_by_classpath, session, caplog + ) -> None: + trigger_runner = TriggerRunner() + mock_trigger_orm = MagicMock() + mock_trigger_orm.kwargs = {"moment": ..., "not_exists_arg": ...} + mock_get_trigger_by_classpath.return_value = {1: mock_trigger_orm} + + trigger_runner.update_triggers({1}) + + assert ( + "Trigger failed; message=__init__() got an unexpected keyword argument 'not_exists_arg'" + in caplog.text + ) + def test_trigger_create_race_condition_18392(session, tmp_path): """