sap_hana: address review feedback on schema collection job

- cancel() now stops the schema collection job too, so its background
  thread and dedicated HANA connection are released promptly on teardown
  instead of lingering until the inactivity timeout.
- Reset the job connection when a HANA error is swallowed inside
  collect_schemas(): the base SchemaCollector catches per-database errors
  and returns, so a transient disconnect never reached run_job's handler
  and the dead connection was reused every cycle. The collector now drops
  its connection reference on HanaError and the job reconnects next cycle.

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

Author: pawel-big-lebowski

sap_hana/datadog_checks/sap_hana/sap_hana.py

@@ -159,9 +159,11 @@ def check(self, _):
                 self._connection_flaked = False
 
     def cancel(self):
-        # Signal the Data Observability async job to stop so its executor thread is
-        # released when the check is unscheduled (e.g. cluster-agent flavor or one-off
-        # check invocations), instead of leaking the DBMAsyncJob thread pool.
+        # Signal both async jobs to stop so their executor threads (and the schema job's
+        # dedicated HANA connection) are released when the check is unscheduled (e.g.
+        # cluster-agent flavor or one-off check invocations), instead of leaking the
+        # DBMAsyncJob thread pool.
+        self._schema_collection_job.cancel()
         self.data_observability.cancel()
 
     def set_default_methods(self):

sap_hana/datadog_checks/sap_hana/schemas.py

@@ -281,21 +281,31 @@ def _get_databases(self) -> list[DatabaseInfo]:
                     pass
                 return [{'name': db_name, 'description': description}]
         except Exception as e:
+            # A dead HANA connection surfaces here rather than propagating, so drop our
+            # reference to signal the owning job to reconnect on the next cycle.
+            if isinstance(e, HanaError):
+                self._conn = None
             self._log.warning("Could not determine current HANA database; skipping schema collection: %s", e)
             return []
 
     @contextlib.contextmanager
     def _get_cursor(self, _database_name):
         conn = self._active_conn()
-        self._query_builder.ensure_stats_permission(conn)
-        query, params = self._query_builder.build()
-        with closing(conn.cursor()) as cursor:
-            cursor.execute(query, params)
-            self._pending_row = cursor.fetchone()
-            try:
-                yield cursor
-            finally:
-                self._pending_row = None
+        try:
+            self._query_builder.ensure_stats_permission(conn)
+            query, params = self._query_builder.build()
+            with closing(conn.cursor()) as cursor:
+                cursor.execute(query, params)
+                self._pending_row = cursor.fetchone()
+                try:
+                    yield cursor
+                finally:
+                    self._pending_row = None
+        except HanaError:
+            # collect_schemas() swallows this per-database error, so signal the dead
+            # connection to the owning job by dropping our reference; it reconnects next cycle.
+            self._conn = None
+            raise
 
     def _get_next(self, cursor):
         """Assemble one table/view from consecutive cursor rows sharing the same (schema, table) key."""
@@ -399,6 +409,15 @@ def run_job(self) -> None:
             self._schema_collector._conn = conn
             self._schema_collector.collect_schemas()
         except HanaError:
-            self._job_conn = None
-            self._schema_collector._conn = None
+            self._reset_conn()
             raise
+        # collect_schemas() swallows per-database HANA errors internally, so a transient
+        # disconnect never reaches the except above. The collector drops its connection
+        # reference in that case; close ours too so the next cycle reconnects instead of
+        # reusing a dead handle forever.
+        if self._schema_collector._conn is None:
+            self._reset_conn()
+
+    def _reset_conn(self) -> None:
+        self._shutdown()
+        self._schema_collector._conn = None

sap_hana/tests/test_schemas_diagnose.py

@@ -230,6 +230,35 @@ def test_schema_job_resets_conn_on_hana_error(self):
         assert job._job_conn is None
         assert job._schema_collector._conn is None
 
+    def test_schema_job_resets_conn_on_swallowed_error(self):
+        # collect_schemas() swallows per-database HANA errors and returns normally after
+        # dropping the collector's connection reference; the job must still reconnect.
+        check = _make_check({'collect_schemas': {'enabled': True, 'run_sync': True}})
+        job = check._schema_collection_job
+        conn = mock.MagicMock()
+        job._job_conn = conn
+
+        def swallow():
+            job._schema_collector._conn = None
+
+        job._schema_collector.collect_schemas = mock.MagicMock(side_effect=swallow)
+
+        job.run_job()
+
+        conn.close.assert_called_once()
+        assert job._job_conn is None
+        assert job._schema_collector._conn is None
+
+    def test_cancel_cancels_schema_collection_job(self):
+        check = _make_check({'collect_schemas': {'enabled': True}})
+        check._schema_collection_job.cancel = mock.MagicMock()
+        check.data_observability.cancel = mock.MagicMock()
+
+        check.cancel()
+
+        check._schema_collection_job.cancel.assert_called_once()
+        check.data_observability.cancel.assert_called_once()
+
     def test_schema_job_disabled_does_not_run(self):
         check = _make_check()
         job = check._schema_collection_job