fix: add default HTTP request timeout to prevent indefinite hangs

Add default timeout of (30s connect, 300s read) to HttpClient.send_request().
When no explicit timeout is provided in request_kwargs, the default is injected
before sending the request. This prevents requests.Session.send() from blocking
indefinitely when a server stalls mid-response (e.g. after a 500 error retry).

ConnectTimeout and ReadTimeout are already in TRANSIENT_EXCEPTIONS, so timeouts
trigger automatic retries with exponential backoff.

Co-Authored-By: alfredo.garcia@airbyte.io <freddy.garcia7.fg@gmail.com>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/streams/http/http_client.py

@@ -85,6 +85,8 @@ def monkey_patched_get_item(self, key):  # type: ignore # this interface is a co
 class HttpClient:
     _DEFAULT_MAX_RETRY: int = 5
     _DEFAULT_MAX_TIME: int = 60 * 10
+    _DEFAULT_CONNECT_TIMEOUT: int = 30
+    _DEFAULT_READ_TIMEOUT: int = 300
     _ACTIONS_TO_RETRY_ON = {
         ResponseAction.RETRY,
         ResponseAction.RATE_LIMITED,
@@ -586,11 +588,17 @@ def send_request(
             verify=request_kwargs.get("verify"),
             cert=request_kwargs.get("cert"),
         )
-        request_kwargs = {**request_kwargs, **env_settings}
+        mutable_request_kwargs: Dict[str, Any] = {**request_kwargs, **env_settings}
+
+        if "timeout" not in mutable_request_kwargs:
+            mutable_request_kwargs["timeout"] = (
+                self._DEFAULT_CONNECT_TIMEOUT,
+                self._DEFAULT_READ_TIMEOUT,
+            )
 
         response: requests.Response = self._send_with_retry(
             request=request,
-            request_kwargs=request_kwargs,
+            request_kwargs=mutable_request_kwargs,
             log_formatter=log_formatter,
             exit_on_rate_limit=exit_on_rate_limit,
         )

unit_tests/sources/streams/http/test_http_client.py

@@ -1059,3 +1059,52 @@ def update_response(*args, **kwargs):
     assert mock_authenticator.access_token == "new_refreshed_token"
     assert returned_response == valid_response
     assert call_count == 2
+
+
+def test_send_request_applies_default_timeout_when_not_provided(mocker):
+    http_client = test_http_client()
+    mocked_response = MagicMock(spec=requests.Response)
+    mocked_response.status_code = 200
+    mocked_response.headers = {}
+    mock_send = mocker.patch.object(requests.Session, "send", return_value=mocked_response)
+
+    http_client.send_request(
+        http_method="get",
+        url="https://test_base_url.com/v1/endpoint",
+        request_kwargs={},
+    )
+
+    assert mock_send.call_count == 1
+    call_kwargs = mock_send.call_args
+    # The timeout should be passed as part of the keyword arguments to session.send()
+    # session.send(request, **request_kwargs) unpacks request_kwargs, so timeout appears as a kwarg
+    assert call_kwargs.kwargs.get("timeout") == (
+        HttpClient._DEFAULT_CONNECT_TIMEOUT,
+        HttpClient._DEFAULT_READ_TIMEOUT,
+    ) or call_kwargs[1].get("timeout") == (
+        HttpClient._DEFAULT_CONNECT_TIMEOUT,
+        HttpClient._DEFAULT_READ_TIMEOUT,
+    )
+
+
+def test_send_request_respects_explicit_timeout(mocker):
+    http_client = test_http_client()
+    mocked_response = MagicMock(spec=requests.Response)
+    mocked_response.status_code = 200
+    mocked_response.headers = {}
+    mock_send = mocker.patch.object(requests.Session, "send", return_value=mocked_response)
+
+    custom_timeout = (10, 60)
+    http_client.send_request(
+        http_method="get",
+        url="https://test_base_url.com/v1/endpoint",
+        request_kwargs={"timeout": custom_timeout},
+    )
+
+    assert mock_send.call_count == 1
+    call_kwargs = mock_send.call_args
+    # The explicit timeout should be preserved, not overridden by the default
+    assert (
+        call_kwargs.kwargs.get("timeout") == custom_timeout
+        or call_kwargs[1].get("timeout") == custom_timeout
+    )