listener: move success hook to after SQLAlchemy commit (#32988)

Signed-off-by: Maciej Obuchowski <obuchowski.maciej@gmail.com>
apache · Aug 1, 2023 · 941c83e · 941c83e
1 parent 17a3dd4
commit 941c83e
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
@@ -1583,11 +1583,13 @@ def _run_raw_task(
             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: