Add TimeoutError to be a retryable error in databricks provider (apac…

…he#43137)

closes apache#43128

providers/src/airflow/providers/databricks/hooks/databricks_base.py

@@ -28,6 +28,7 @@
 import copy
 import platform
 import time
+from asyncio.exceptions import TimeoutError
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
 from urllib.parse import urlsplit
@@ -679,7 +680,7 @@ def _retryable_error(exception: BaseException) -> bool:
             if exception.status >= 500 or exception.status == 429:
                 return True
 
-        if isinstance(exception, ClientConnectorError):
+        if isinstance(exception, (ClientConnectorError, TimeoutError)):
             return True
 
         return False

providers/tests/databricks/hooks/test_databricks.py

@@ -21,6 +21,7 @@
 import json
 import ssl
 import time
+from asyncio.exceptions import TimeoutError
 from unittest import mock
 from unittest.mock import AsyncMock
 
@@ -1551,6 +1552,16 @@ async def test_do_api_call_retries_with_client_connector_error(self, mock_get):
                     await self.hook._a_do_api_call(GET_RUN_ENDPOINT, {})
                 assert mock_errors.call_count == DEFAULT_RETRY_NUMBER
 
+    @pytest.mark.asyncio
+    @mock.patch("airflow.providers.databricks.hooks.databricks_base.aiohttp.ClientSession.get")
+    async def test_do_api_call_retries_with_client_timeout_error(self, mock_get):
+        mock_get.side_effect = TimeoutError()
+        with mock.patch.object(self.hook.log, "error") as mock_errors:
+            async with self.hook:
+                with pytest.raises(AirflowException):
+                    await self.hook._a_do_api_call(GET_RUN_ENDPOINT, {})
+                assert mock_errors.call_count == DEFAULT_RETRY_NUMBER
+
     @pytest.mark.asyncio
     @mock.patch("airflow.providers.databricks.hooks.databricks_base.aiohttp.ClientSession.get")
     async def test_do_api_call_retries_with_retryable_error(self, mock_get):