Suppress jaydebeapi.Error when setAutoCommit or getAutoCommit is unsupported by JDBC driver

airflow/providers/jdbc/hooks/jdbc.py

@@ -22,6 +22,7 @@
 import jaydebeapi
 
 from airflow.providers.common.sql.hooks.sql import DbApiHook
+from airflow.utils.context import suppress_and_warn
 
 if TYPE_CHECKING:
     from airflow.models.connection import Connection
@@ -152,7 +153,8 @@ def set_autocommit(self, conn: jaydebeapi.Connection, autocommit: bool) -> None:
         :param conn: The connection.
         :param autocommit: The connection's autocommit setting.
         """
-        conn.jconn.setAutoCommit(autocommit)
+        with suppress_and_warn(jaydebeapi.Error):
+            conn.jconn.setAutoCommit(autocommit)
 
     def get_autocommit(self, conn: jaydebeapi.Connection) -> bool:
         """Get autocommit setting for the provided connection.
@@ -162,4 +164,5 @@ def get_autocommit(self, conn: jaydebeapi.Connection) -> bool:
             to True on the connection. False if it is either not set, set to
             False, or the connection does not support auto-commit.
         """
-        return conn.jconn.getAutoCommit()
+        with suppress_and_warn(jaydebeapi.Error):
+            return conn.jconn.getAutoCommit()

airflow/utils/context.py

@@ -22,6 +22,7 @@
 import contextlib
 import copy
 import functools
+import traceback
 import warnings
 from typing import (
     TYPE_CHECKING,
@@ -357,6 +358,15 @@ def _create_value(k: str, v: Any) -> Any:
     return {k: _create_value(k, v) for k, v in source._context.items()}
 
 
+@contextlib.contextmanager
+def suppress_and_warn(*exceptions: type[BaseException]):
+    """Context manager that suppresses the given exceptions and logs a warning message."""
+    try:
+        yield
+    except exceptions as e:
+        warnings.warn(f"Exception suppressed: {e}\n{traceback.format_exc()}", category=UserWarning)
+
+
 def context_get_dataset_events(context: Context) -> DatasetEventAccessors:
     try:
         return context["dataset_events"]

airflow/utils/context.pyi

@@ -127,4 +127,5 @@ def context_merge(context: Context, **kwargs: Any) -> None: ...
 def context_update_for_unmapped(context: Context, task: BaseOperator) -> None: ...
 def context_copy_partial(source: Context, keys: Container[str]) -> Context: ...
 def lazy_mapping_from_context(source: Context) -> Mapping[str, Any]: ...
+def suppress_and_warn(*exceptions: type[BaseException]): ...
 def context_get_dataset_events(context: Context) -> DatasetEventAccessors: ...

tests/providers/jdbc/hooks/test_jdbc.py

@@ -22,6 +22,7 @@
 from unittest import mock
 from unittest.mock import Mock, patch
 
+import jaydebeapi
 import pytest
 
 from airflow.models import Connection
@@ -82,13 +83,27 @@ def test_jdbc_conn_set_autocommit(self, _):
         jdbc_hook.set_autocommit(jdbc_conn, False)
         jdbc_conn.jconn.setAutoCommit.assert_called_once_with(False)
 
+    @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
+    def test_jdbc_conn_set_autocommit_when_not_supported(self, _):
+        jdbc_hook = JdbcHook()
+        jdbc_conn = jdbc_hook.get_conn()
+        jdbc_conn.jconn.setAutoCommit.side_effect = jaydebeapi.Error()
+        jdbc_hook.set_autocommit(jdbc_conn, False)
+
     @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
     def test_jdbc_conn_get_autocommit(self, _):
         jdbc_hook = JdbcHook()
         jdbc_conn = jdbc_hook.get_conn()
         jdbc_hook.get_autocommit(jdbc_conn)
         jdbc_conn.jconn.getAutoCommit.assert_called_once_with()
 
+    @patch("airflow.providers.jdbc.hooks.jdbc.jaydebeapi.connect")
+    def test_jdbc_conn_get_autocommit_when_not_supported_then_return_false(self, _):
+        jdbc_hook = JdbcHook()
+        jdbc_conn = jdbc_hook.get_conn()
+        jdbc_conn.jconn.getAutoCommit.side_effect = jaydebeapi.Error()
+        assert jdbc_hook.get_autocommit(jdbc_conn) is False
+
     def test_driver_hook_params(self):
         hook = get_hook(hook_params=dict(driver_path="Blah driver path", driver_class="Blah driver class"))
         assert hook.driver_path == "Blah driver path"

tests/utils/test_context.py

@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import pytest
+
+from airflow.exceptions import DeserializingResultError
+from airflow.utils.context import AirflowContextDeprecationWarning, suppress_and_warn
+
+
+def method_which_raises_an_airflow_deprecation_warning():
+    raise AirflowContextDeprecationWarning()
+
+
+class TestContext:
+    def test_suppress_and_warn_when_raised_exception_is_suppressed(self):
+        with suppress_and_warn(AirflowContextDeprecationWarning):
+            method_which_raises_an_airflow_deprecation_warning()
+
+    def test_suppress_and_warn_when_raised_exception_is_not_suppressed(self):
+        with pytest.raises(AirflowContextDeprecationWarning):
+            with suppress_and_warn(DeserializingResultError):
+                method_which_raises_an_airflow_deprecation_warning()