fix: attribute recovered-kernel connection failures to the kernel for telemetry

The connection-failure telemetry suppression read the original connect() kwargs
to decide whether the failed connection was a kernel connection. On the Reyden
auto-recovery path the kernel retry uses a kwargs copy, so the original still
said Thrift — a kernel open-failure was logged by the wrapper despite the kernel
owning telemetry for kernel connections. Decide from the session that actually
failed (self.session.use_kernel) instead. If the kernel was never constructed
(e.g. its wheel is missing), self.session stays Thrift and the wrapper still
logs, so that otherwise-invisible failure is still recorded.

Co-authored-by: Isaac <no-reply@databricks.com>
Signed-off-by: Rahul Singhal <rahul.singhal@databricks.com>

Author: rahuls-db

src/databricks/sql/client.py

@@ -417,10 +417,18 @@ def read(self) -> Optional[OAuthToken]:
             )
         except Exception as e:
             # Respect user's telemetry preference even during connection failure.
-            # For use_kernel connections the kernel owns telemetry, so suppress
-            # the wrapper-side failure log to avoid wrapper-vs-kernel duplication.
-            enable_telemetry = kwargs.get("enable_telemetry", True) and not kwargs.get(
-                "use_kernel", False
+            # For a kernel connection the kernel owns telemetry, so suppress the
+            # wrapper-side failure log to avoid wrapper-vs-kernel duplication.
+            # Read the backend from the session that actually failed rather than
+            # the caller's kwargs: on the Reyden auto-recovery path we retry on
+            # the kernel via a kwargs copy, so the original kwargs still says
+            # Thrift. If the kernel never got constructed (e.g. its wheel is
+            # missing), self.session is the Thrift session and we still log.
+            attempted_kernel = getattr(
+                getattr(self, "session", None), "use_kernel", False
+            )
+            enable_telemetry = (
+                kwargs.get("enable_telemetry", True) and not attempted_kernel
             )
             TelemetryClientFactory.connection_failure_log(
                 error_name="Exception",

tests/unit/test_session.py

@@ -910,3 +910,23 @@ def test_kernel_retry_failure_chains_both_errors(self, mock_thrift):
         # Kernel failure is surfaced as primary; the Thrift rejection is
         # preserved in the chain for diagnosis.
         assert isinstance(excinfo.value.__cause__, ReydenThriftUnsupportedError)
+
+    @patch("%s.session.ThriftDatabricksClient" % PACKAGE)
+    def test_recovered_kernel_failure_suppresses_wrapper_telemetry(self, mock_thrift):
+        # A connection that recovered onto the kernel and then failed there must
+        # NOT emit the wrapper's connection-failure log — the kernel owns
+        # telemetry for kernel connections. Guards against reading the original
+        # (Thrift) kwargs instead of the session that actually failed.
+        mock_thrift.return_value.open_session.side_effect = self._reject()
+        with self._fake_kernel() as mock_kernel, patch(
+            "databricks.sql.client.TelemetryClientFactory.connection_failure_log"
+        ) as mock_fail_log:
+            mock_kernel.return_value.open_session.side_effect = OperationalError(
+                "kernel boom"
+            )
+            # enable_telemetry=True so only the kernel-suppression logic can flip
+            # it off — proving the fix rather than the user's opt-out.
+            with pytest.raises(OperationalError):
+                self._connect(enable_telemetry=True)
+        mock_fail_log.assert_called_once()
+        assert mock_fail_log.call_args.kwargs["enable_telemetry"] is False