fix(ingestion): keep connection_obj in sync with engine across DB switches

self.connection_obj is set once in __init__ to the initial engine and
never updated. After set_inspector rebuilds self.engine, connection_obj
still points at the disposed original engine — pinning its dialect and
compiled_cache alive for the source's lifetime.

Rebind connection_obj when creating the new engine in set_inspector,
and clear it in _release_engine so close() leaves nothing dangling.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ulixius9

ingestion/src/metadata/ingestion/source/database/common_db_source.py

@@ -158,6 +158,7 @@ def set_inspector(self, database_name: str) -> None:
         new_service_connection.database = database_name
         self.engine = get_connection(new_service_connection)
         self.session = create_and_bind_thread_safe_session(self.engine)
+        self.connection_obj = self.engine
 
     def _release_engine(self) -> None:
         # Close fairies first so _ConnectionRecord drops its pool reference;
@@ -185,6 +186,7 @@ def _release_engine(self) -> None:
         except Exception as exc:  # pylint: disable=broad-except
             logger.warning(f"Failed to dispose engine: {exc}")
         self.engine = None
+        self.connection_obj = None
 
     def get_database_names(self) -> Iterable[str]:
         """

ingestion/tests/unit/topology/database/test_common_db_source.py

@@ -353,6 +353,7 @@ class _ReleaseOnlySurrogate(CommonDbSourceService):
 
     def __init__(self, engine=None):  # pylint: disable=super-init-not-called
         self.engine = engine
+        self.connection_obj = engine
         self._connection_map = {}
         self._inspector_map = {}
         self.session = None
@@ -454,6 +455,16 @@ def test_tolerates_already_closed_connection(self, surrogate):
         assert healthy.closed is True
         assert surrogate._connection_map == {}
 
+    def test_clears_connection_obj_alongside_engine(self, surrogate):
+        # connection_obj is set in __init__ to the initial engine and used by
+        # test_connection(); without clearing it on release, it pins the
+        # original Engine alive for the source's lifetime even after dispose.
+        assert surrogate.connection_obj is surrogate.engine
+
+        surrogate._release_engine()
+
+        assert surrogate.connection_obj is None
+
     def test_closes_connections_from_arbitrary_thread_ids(self, surrogate):
         """Key property of Option B: close-all, not detach-current-thread.
         Every fairy in _connection_map must close regardless of the caller's