From e565635e858dccdf9e57a5e82ed79a970df1c3bc Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Mon, 19 Jun 2023 17:14:48 +0800 Subject: [PATCH 1/6] fix(jobs/triggerer_job_runner): catch the exception that triggerer initialization failed --- airflow/jobs/triggerer_job_runner.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 8719f4ea95c8f..fa64adf01eed0 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -680,7 +680,13 @@ 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 From d0f53d9a1f344f158eaaaf0d4e00fbc47eb315d0 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Mon, 19 Jun 2023 18:45:19 +0800 Subject: [PATCH 2/6] test(jobs/triggerer_job_runner): add test case for checking TriggerRunner.update_trigger catch triggerer change exception --- airflow/jobs/triggerer_job_runner.py | 1 + tests/jobs/test_triggerer_job.py | 22 +++++++++++++++++++++- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index fa64adf01eed0..79b5e953dcd49 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -680,6 +680,7 @@ 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 + try: new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) except TypeError as err: 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): """ From f54b13e78fa8b5550fc9748cedd8286125ba1912 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Tue, 20 Jun 2023 10:40:28 +0800 Subject: [PATCH 3/6] refactor(jobs/triggerer_job_runner): catch multiple exception cases in the same try catch block when initialing trigger --- airflow/jobs/triggerer_job_runner.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 79b5e953dcd49..60ff26681fe05 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -676,14 +676,10 @@ def update_triggers(self, requested_trigger_ids: set[int]): try: new_trigger_orm = new_triggers[new_id] trigger_class = self.get_trigger_by_classpath(new_trigger_orm.classpath) - except BaseException as e: - # Either the trigger code or the path to it is bad. Fail the trigger. - self.failed_triggers.append((new_id, e)) - continue - - try: new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) - except TypeError as err: + except (BaseException, TypeError) as err: + # BaseException: Either the trigger code or the path to it is bad. Fail the trigger. + # TypeError: The argument of the __init__ method in the trigger might have been changed. self.log.error("Trigger failed; message=%s", err) self.failed_triggers.append((new_id, err)) continue From fb5d9b61434469641f283bef6a46726c1c5f9999 Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Tue, 20 Jun 2023 15:57:44 +0800 Subject: [PATCH 4/6] refactor(jobs/triggerer_job_runner): remove unnecessary TypeError catching Co-authored-by: Tzu-ping Chung --- airflow/jobs/triggerer_job_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 60ff26681fe05..1da855822a635 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -677,7 +677,7 @@ def update_triggers(self, requested_trigger_ids: set[int]): new_trigger_orm = new_triggers[new_id] trigger_class = self.get_trigger_by_classpath(new_trigger_orm.classpath) new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) - except (BaseException, TypeError) as err: + except BaseException as err: # BaseException: Either the trigger code or the path to it is bad. Fail the trigger. # TypeError: The argument of the __init__ method in the trigger might have been changed. self.log.error("Trigger failed; message=%s", err) From 35458de8116bfcb13a88c6da17463dba0704d7ef Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Tue, 20 Jun 2023 22:15:36 +0800 Subject: [PATCH 5/6] Revert "refactor(jobs/triggerer_job_runner): remove unnecessary TypeError catching" This reverts commit ac561a2416f16a8f9a3b211715fd4a7e012b47ad. --- airflow/jobs/triggerer_job_runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 1da855822a635..60ff26681fe05 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -677,7 +677,7 @@ def update_triggers(self, requested_trigger_ids: set[int]): new_trigger_orm = new_triggers[new_id] trigger_class = self.get_trigger_by_classpath(new_trigger_orm.classpath) new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) - except BaseException as err: + except (BaseException, TypeError) as err: # BaseException: Either the trigger code or the path to it is bad. Fail the trigger. # TypeError: The argument of the __init__ method in the trigger might have been changed. self.log.error("Trigger failed; message=%s", err) From 5a126531d296795f39e12b32d131c7816f92563d Mon Sep 17 00:00:00 2001 From: Wei Lee Date: Tue, 20 Jun 2023 22:15:40 +0800 Subject: [PATCH 6/6] Revert "refactor(jobs/triggerer_job_runner): catch multiple exception cases in the same try catch block when initialing trigger" This reverts commit 4f3b24b8a7698973fe319618b6ff9f459dc0155a. --- airflow/jobs/triggerer_job_runner.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/airflow/jobs/triggerer_job_runner.py b/airflow/jobs/triggerer_job_runner.py index 60ff26681fe05..79b5e953dcd49 100644 --- a/airflow/jobs/triggerer_job_runner.py +++ b/airflow/jobs/triggerer_job_runner.py @@ -676,10 +676,14 @@ def update_triggers(self, requested_trigger_ids: set[int]): try: new_trigger_orm = new_triggers[new_id] trigger_class = self.get_trigger_by_classpath(new_trigger_orm.classpath) + except BaseException as e: + # Either the trigger code or the path to it is bad. Fail the trigger. + self.failed_triggers.append((new_id, e)) + continue + + try: new_trigger_instance = trigger_class(**new_trigger_orm.kwargs) - except (BaseException, TypeError) as err: - # BaseException: Either the trigger code or the path to it is bad. Fail the trigger. - # TypeError: The argument of the __init__ method in the trigger might have been changed. + except TypeError as err: self.log.error("Trigger failed; message=%s", err) self.failed_triggers.append((new_id, err)) continue