Skip to content

Catch the exception that triggerer initialization failed #31999

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hussein-awala merged 6 commits into from
Jun 20, 2023
Merged

Catch the exception that triggerer initialization failed #31999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e565635
fix(jobs/triggerer_job_runner): catch the exception that triggerer in…
Lee-W Jun 19, 2023
d0f53d9
test(jobs/triggerer_job_runner): add test case for checking TriggerRu…
Lee-W Jun 19, 2023
f54b13e
refactor(jobs/triggerer_job_runner): catch multiple exception cases i…
Lee-W Jun 20, 2023
fb5d9b6
refactor(jobs/triggerer_job_runner): remove unnecessary TypeError cat…
Lee-W Jun 20, 2023
35458de
Revert "refactor(jobs/triggerer_job_runner): remove unnecessary TypeE…
Lee-W Jun 20, 2023
5a12653
Revert "refactor(jobs/triggerer_job_runner): catch multiple exception…
Lee-W Jun 20, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion airflow/jobs/triggerer_job_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@uranusjr uranusjr Jun 20, 2023 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like this is the only meaningful change (and the log). Maybe better to do it like this?

try:
    ...
except BaseException as err:
    ...
try:
    new_trigger_instance = trigger_class(**new_trigger_orm.kwargs)
except TypeError as err:
    ...

@Lee-W Lee-W Jun 20, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the most straightforward way of what exceptions might happen in which part of the code. The first thought I had when I received the previous comment was whether it's difficult to read to have so many try-catch blocks, and their error handling is almost the same. That's why I change it to the current implementation.

Another implementation I'm thinking of is

            try:
                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 TypeError as err:
                # 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, e))
            except BaseException as err:
                # Either the trigger code or the path to it is bad. Fail the trigger.
                self.failed_triggers.append((new_id, err))

But we still need to handle the failed_triggeres on both sections in this implementation. The only benefit is that we reduce one try-catch block.

Please let me know which is preferred. Thanks!

@pankajkoti pankajkoti Jun 20, 2023 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like the approach suggested in #31999 (comment) by @uranusjr as it states explicitly what errors would be raised and in which steps. I agree the exception handling is same in both cases but the approach suggested gives more clarity actually when reading the code.
No strong opinion though :)

@Lee-W Lee-W Jun 20, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After a second though, it's indeed more clear. let me revert back to that version 🙂

@hussein-awala hussein-awala Jun 20, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that there is no big difference between the two approaches since we will get different error messages


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
Expand Down
22 changes: 21 additions & 1 deletion tests/jobs/test_triggerer_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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):
"""
Expand Down