[airflow] branch listener-move-complete created (now 09098ec749)

[mo...@apache.org]

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

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


      at 09098ec749 listener: move success hook to after SQLAlchemy commit

This branch includes the following new commits:

     new 09098ec749 listener: move success hook to after SQLAlchemy commit

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.

[airflow] 01/01: listener: move success hook to after SQLAlchemy commit

[mo...@apache.org]

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

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

commit 09098ec7496deca3f89a8f04aca3f0b342634d76
Author: Maciej Obuchowski <ob...@gmail.com>
AuthorDate: Mon Jul 31 20:20:43 2023 +0200

    listener: move success hook to after SQLAlchemy commit
    
    Signed-off-by: Maciej Obuchowski <ob...@gmail.com>
---
 airflow/models/taskinstance.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index c31e2d8c15..37d5da21d0 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -1583,11 +1583,13 @@ class TaskInstance(Base, LoggingMixin):
             session.merge(self).task = self.task
             if self.state == TaskInstanceState.SUCCESS:
                 self._register_dataset_changes(session=session)
+
+            session.commit()
+            if self.state == TaskInstanceState.SUCCESS:
                 get_listener_manager().hook.on_task_instance_success(
                     previous_state=TaskInstanceState.RUNNING, task_instance=self, session=session
                 )
 
-            session.commit()
         return None
 
     def _register_dataset_changes(self, *, session: Session) -> None: