Rollback the connection after server ping method

In the method ``engines._connect_ping_listener``, the connection should
be rolled back after the ping execution. The rollback will revert the
transaction and delete it.

Closes-Bug: #2008209
Change-Id: Iba29ded227634e02795052acfd89b572bf21f54c

oslo_db/sqlalchemy/engines.py

@@ -81,9 +81,17 @@ def _connect_ping_listener(connection, branch):
         # run the select again to re-validate the Connection.
         LOG.exception(
             'Database connection was found disconnected; reconnecting')
+        # TODO(ralonsoh): drop this attr check once SQLAlchemy minimum version
+        # is 2.0.
+        if hasattr(connection, 'rollback'):
+            connection.rollback()
         connection.scalar(select(1))
     finally:
         connection.should_close_with_result = save_should_close_with_result
+        # TODO(ralonsoh): drop this attr check once SQLAlchemy minimum version
+        # is 2.0.
+        if hasattr(connection, 'rollback'):
+            connection.rollback()
 
 
 def _setup_logging(connection_debug=0):

oslo_db/tests/sqlalchemy/test_sqlalchemy.py

@@ -23,7 +23,9 @@ from unittest import mock
 
 import fixtures
 from oslo_config import cfg
+from oslo_utils import versionutils
 import sqlalchemy
+from sqlalchemy.engine import base as base_engine
 from sqlalchemy import exc
 from sqlalchemy import sql
 from sqlalchemy import Column, MetaData, Table
@@ -895,3 +897,22 @@ class PatchStacktraceTest(db_test_base._DbTestCase):
             # we're the caller, see that we're in there
             caller = os.path.join("tests", "sqlalchemy", "test_sqlalchemy.py")
             self.assertIn(caller, call[1][1])
+
+
+class MySQLConnectPingListenerTest(db_test_base._MySQLOpportunisticTestCase):
+
+    def test__connect_ping_listener(self):
+        for idx in range(2):
+            with self.engine.begin() as conn:
+                self.assertTrue(isinstance(conn._transaction,
+                                           base_engine.RootTransaction))
+                engines._connect_ping_listener(conn, False)
+                # TODO(ralonsoh): drop this check once SQLAlchemy minimum
+                #  version is 2.0.
+                sqla_version = versionutils.convert_version_to_tuple(
+                    sqlalchemy.__version__)
+                if sqla_version[0] >= 2:
+                    self.assertIsNone(conn._transaction)
+                else:
+                    self.assertTrue(isinstance(conn._transaction,
+                                               base_engine.RootTransaction))