[airflow] branch main updated: Fix "This Session's transaction has been rolled back" (#25532)

[ep...@apache.org]

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

ephraimanierobi 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 5668888a7e Fix "This Session's transaction has been rolled back" (#25532)
5668888a7e is described below

commit 5668888a7e1074a620b3d38f407ecf1aa055b623
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Fri Aug 5 17:15:31 2022 +0100

    Fix "This Session's transaction has been rolled back" (#25532)
    
    Accessing the run_id(self.run_id) on exception leads to error because sessions are invalidated on exception. Here we extract the run_id before handling the exception
---
 airflow/models/dagrun.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index e902f687a0..c3ac204dfe 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -1138,6 +1138,10 @@ class DagRun(Base, LoggingMixin):
         :param session: the session to use
 
         """
+        # Fetch the information we need before handling the exception to avoid
+        # PendingRollbackError due to the session being invalidated on exception
+        # see https://github.com/apache/superset/pull/530
+        run_id = self.run_id
         try:
             if hook_is_noop:
                 session.bulk_insert_mappings(TI, tasks)
@@ -1151,7 +1155,7 @@ class DagRun(Base, LoggingMixin):
             self.log.info(
                 'Hit IntegrityError while creating the TIs for %s- %s',
                 dag_id,
-                self.run_id,
+                run_id,
                 exc_info=True,
             )
             self.log.info('Doing session rollback.')