[airflow] branch listener-move-onrunning-callback updated (983dcc95ee -> 3c7ba730f2)

[mo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mobuchowski pushed a change to branch listener-move-onrunning-callback
in repository https://gitbox.apache.org/repos/asf/airflow.git


    omit 983dcc95ee listener: call on_task_instance_running after rendering templates
     new 3c7ba730f2 listener: call on_task_instance_running after rendering templates

This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version.  This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:

 * -- * -- B -- O -- O -- O   (983dcc95ee)
            \
             N -- N -- N   refs/heads/listener-move-onrunning-callback (3c7ba730f2)

You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.

Any revisions marked "omit" are not gone; other references still
refer to them.  Any revisions marked "discard" are gone forever.

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.


Summary of changes:
 airflow/models/taskinstance.py | 11 +++--------
 1 file changed, 3 insertions(+), 8 deletions(-)

[airflow] 01/01: listener: call on_task_instance_running after rendering templates

[mo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mobuchowski pushed a commit to branch listener-move-onrunning-callback
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 3c7ba730f2ebb4b3509fea88be21215ea3c99a05
Author: Maciej Obuchowski <ob...@gmail.com>
AuthorDate: Thu Jul 20 14:20:00 2023 +0200

    listener: call on_task_instance_running after rendering templates
    
    Signed-off-by: Maciej Obuchowski <ob...@gmail.com>
---
 airflow/models/taskinstance.py | 21 ++++++++++++---------
 1 file changed, 12 insertions(+), 9 deletions(-)

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index 4abfd94cd0..5649d26fe3 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1498,14 +1498,9 @@ class TaskInstance(Base, LoggingMixin):
         self.task = self.task.prepare_for_execution()
         context = self.get_template_context(ignore_param_exceptions=False)
 
-        # We lose previous state because it's changed in other process in LocalTaskJob.
-        # We could probably pass it through here though...
-        get_listener_manager().hook.on_task_instance_running(
-            previous_state=TaskInstanceState.QUEUED, task_instance=self, session=session
-        )
         try:
             if not mark_success:
-                self._execute_task_with_callbacks(context, test_mode)
+                self._execute_task_with_callbacks(context, test_mode, session)
             if not test_mode:
                 self.refresh_from_db(lock_for_update=True, session=session)
             self.state = TaskInstanceState.SUCCESS
@@ -1601,7 +1596,8 @@ class TaskInstance(Base, LoggingMixin):
                     session=session,
                 )
 
-    def _execute_task_with_callbacks(self, context, test_mode=False):
+    @provide_session
+    def _execute_task_with_callbacks(self, context, test_mode: bool = False, session: Session = NEW_SESSION):
         """Prepare Task for Execution."""
         from airflow.models.renderedtifields import RenderedTaskInstanceFields
 
@@ -1651,16 +1647,23 @@ class TaskInstance(Base, LoggingMixin):
                 )
 
             # Run pre_execute callback
-            self.task.pre_execute(context=context)
+            # Is never MappedOperator at this point
+            self.task.pre_execute(context=context)  # type: ignore[union-attr]
 
             # Run on_execute callback
             self._run_execute_callback(context, self.task)
 
+            # Run on_task_instance_running event
+            get_listener_manager().hook.on_task_instance_running(
+                previous_state=TaskInstanceState.QUEUED, task_instance=self, session=session
+            )
+
             # Execute the task
             with set_current_context(context):
                 result = self._execute_task(context, task_orig)
             # Run post_execute callback
-            self.task.post_execute(context=context, result=result)
+            # Is never MappedOperator at this point
+            self.task.post_execute(context=context, result=result)  # type: ignore[union-attr]
 
         Stats.incr(f"operator_successes_{self.task.task_type}", tags=self.stats_tags)
         # Same metric with tagging