diff --git a/oslo_db/sqlalchemy/exc_filters.py b/oslo_db/sqlalchemy/exc_filters.py
index 2f575d84..9803343c 100644
--- a/oslo_db/sqlalchemy/exc_filters.py
+++ b/oslo_db/sqlalchemy/exc_filters.py
@@ -485,7 +485,11 @@ def register_engine(engine):
     def rollback_savepoint(conn, name, context):
         exc_info = sys.exc_info()
         if exc_info[1]:
-            conn.info[ROLLBACK_CAUSE_KEY] = exc_info[1]
+            # NOTE(zzzeek) accessing conn.info on an invalidated
+            # connection causes it to reconnect, which we don't
+            # want to do inside a rollback handler
+            if not conn.invalidated:
+                conn.info[ROLLBACK_CAUSE_KEY] = exc_info[1]
         # NOTE(zzzeek) this eliminates a reference cycle between tracebacks
         # that would occur in Python 3 only, which has been shown to occur if
         # this function were in fact part of the traceback.  That's not the
@@ -497,7 +501,11 @@ def register_engine(engine):
     @event.listens_for(engine, "rollback")
     @event.listens_for(engine, "commit")
     def pop_exc_tx(conn):
-        conn.info.pop(ROLLBACK_CAUSE_KEY, None)
+        # NOTE(zzzeek) accessing conn.info on an invalidated
+        # connection causes it to reconnect, which we don't
+        # want to do inside a rollback handler
+        if not conn.invalidated:
+            conn.info.pop(ROLLBACK_CAUSE_KEY, None)
 
     # .. as well as connection pool checkin (just in case).
     # the .info dictionary lasts as long as the DBAPI connection itself
diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py
index 558cca8d..ce148db6 100644
--- a/oslo_db/tests/sqlalchemy/test_exc_filters.py
+++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py
@@ -712,6 +712,55 @@ class TestExceptionCauseMySQLSavepoint(test_base.MySQLOpportunisticTestCase):
         except exception.DBError as dbe_outer:
             self.assertIsNone(dbe_outer.cause)
 
+    def test_rollback_doesnt_interfere_with_killed_conn(self):
+        session = self.sessionmaker()
+
+        session.begin()
+        try:
+            session.execute("select 1")
+
+            # close underying DB connection
+            session.connection().connection.connection.close()
+
+            # alternate approach, but same idea:
+            # conn_id = session.scalar("select connection_id()")
+            # session.execute("kill connection %s" % conn_id)
+
+            # try using it, will raise an error
+            session.execute("select 1")
+        except exception.DBConnectionError:
+            # issue being tested is that this session.rollback()
+            # does not itself try to re-connect and raise another
+            # error.
+            session.rollback()
+        else:
+            assert False, "no exception raised"
+
+    def test_savepoint_rollback_doesnt_interfere_with_killed_conn(self):
+        session = self.sessionmaker()
+
+        session.begin()
+        try:
+            session.begin_nested()
+            session.execute("select 1")
+
+            # close underying DB connection
+            session.connection().connection.connection.close()
+
+            # alternate approach, but same idea:
+            # conn_id = session.scalar("select connection_id()")
+            # session.execute("kill connection %s" % conn_id)
+
+            # try using it, will raise an error
+            session.execute("select 1")
+        except exception.DBConnectionError:
+            # issue being tested is that this session.rollback()
+            # does not itself try to re-connect and raise another
+            # error.
+            session.rollback()
+        else:
+            assert False, "no exception raised"
+
 
 class TestDBDataErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase):