This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new ad30ba1284 Call listener on_task_instance_failed() after ti state is
changed (#41053)
ad30ba1284 is described below
commit ad30ba128443c72e3805bd77932f1fdc47bdd7f5
Author: Kacper Muda <[email protected]>
AuthorDate: Thu Aug 1 13:51:12 2024 +0200
Call listener on_task_instance_failed() after ti state is changed (#41053)
Signed-off-by: Kacper Muda <[email protected]>
---
airflow/models/taskinstance.py | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 9210659d4c..7098c53d69 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -3279,10 +3279,6 @@ class TaskInstance(Base, LoggingMixin):
:param fail_stop: if true, stop remaining tasks in dag
"""
- get_listener_manager().hook.on_task_instance_failed(
- previous_state=TaskInstanceState.RUNNING, task_instance=ti,
error=error, session=session
- )
-
if error:
if isinstance(error, BaseException):
tb = TaskInstance.get_truncated_error_traceback(error,
truncate_to=ti._execute_task)
@@ -3356,6 +3352,10 @@ class TaskInstance(Base, LoggingMixin):
email_for_state = operator.attrgetter("email_on_retry")
callbacks = task.on_retry_callback if task else None
+ get_listener_manager().hook.on_task_instance_failed(
+ previous_state=TaskInstanceState.RUNNING, task_instance=ti,
error=error, session=session
+ )
+
return {
"ti": ti,
"email_for_state": email_for_state,
