You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by mo...@apache.org on 2024/08/01 11:51:19 UTC
(airflow) branch main updated: Call listener on_task_instance_failed() after ti state is changed (#41053)
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 <mu...@gmail.com>
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 <mu...@gmail.com>
---
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,