feat(low-code): opt-in TCP keepalive and request timeouts on HttpRequester

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2778,6 +2778,27 @@ definitions:
         description: Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).
         type: boolean
         default: false
+      connect_timeout_in_seconds:
+        title: Connect Timeout (seconds)
+        description: Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied.
+        type: number
+        exclusiveMinimum: 0
+        examples:
+          - 30
+      read_timeout_in_seconds:
+        title: Read Timeout (seconds)
+        description: The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries.
+        type: number
+        exclusiveMinimum: 0
+        examples:
+          - 300
+      use_tcp_keepalive:
+        title: Use TCP Keepalive
+        description: Enables TCP keepalive (60 s idle, 10 s probe interval, 6 probes) on this requester's connections, so dead or NAT-dropped connections are detected and fail instead of hanging forever. Does not fail slow but alive servers; use read_timeout_in_seconds to bound those.
+        type: boolean
+        default: false
+        examples:
+          - true
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -3204,6 +3204,26 @@ class HttpRequester(BaseModelWithDeprecations):
         description="Enables stream requests caching. When set to true, repeated requests to the same URL will return cached responses. Parent streams automatically have caching enabled. Only set this to false if you are certain that caching should be disabled, as it may negatively impact performance when the same data is needed multiple times (e.g., for scroll-based pagination APIs where caching causes duplicate records).",
         title="Use Cache",
     )
+    connect_timeout_in_seconds: Optional[float] = Field(
+        None,
+        description="Maximum time to establish the connection (TCP + TLS handshake). When unset, no connect timeout is applied.",
+        examples=[30],
+        gt=0,
+        title="Connect Timeout (seconds)",
+    )
+    read_timeout_in_seconds: Optional[float] = Field(
+        None,
+        description="The maximum time to wait for the server to send any data (the first byte of the response, or the next chunk while the body is read). It is NOT a limit on the total request duration. When unset, no read timeout is applied. Only set it when every endpoint served by this requester responds well within the value; synchronous report or export endpoints can stay silent for minutes. Each timed-out attempt counts against the error handler's retry window (600 s for manifest connectors), so a value near 300 allows only a few retries.",
+        examples=[300],
+        gt=0,
+        title="Read Timeout (seconds)",
+    )
+    use_tcp_keepalive: Optional[bool] = Field(
+        False,
+        description="Enables TCP keepalive (60 s idle, 10 s probe interval, 6 probes) on this requester's connections, so dead or NAT-dropped connections are detected and fail instead of hanging forever. Does not fail slow but alive servers; use read_timeout_in_seconds to bound those.",
+        examples=[True],
+        title="Use TCP Keepalive",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2803,6 +2803,9 @@ def create_http_requester(
             parameters=model.parameters or {},
             message_repository=self._message_repository,
             use_cache=should_use_cache,
+            connect_timeout_in_seconds=model.connect_timeout_in_seconds,
+            read_timeout_in_seconds=model.read_timeout_in_seconds,
+            use_tcp_keepalive=model.use_tcp_keepalive or False,
             decoder=decoder,
             stream_response=decoder.is_stream_response() if decoder else False,
         )

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -66,6 +66,9 @@ class HttpRequester(Requester):
     disable_retries: bool = False
     message_repository: MessageRepository = NoopMessageRepository()
     use_cache: bool = False
+    connect_timeout_in_seconds: Optional[float] = None
+    read_timeout_in_seconds: Optional[float] = None
+    use_tcp_keepalive: bool = False
     _exit_on_rate_limit: bool = False
     stream_response: bool = False
     decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={}))
@@ -114,6 +117,13 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
             backoff_strategy=backoff_strategies,
             disable_retries=self.disable_retries,
             message_repository=self.message_repository,
+            request_timeout=(
+                (self.connect_timeout_in_seconds, self.read_timeout_in_seconds)
+                if self.connect_timeout_in_seconds is not None
+                or self.read_timeout_in_seconds is not None
+                else None
+            ),
+            use_tcp_keepalive=self.use_tcp_keepalive,
         )
 
     @property

airbyte_cdk/sources/streams/http/http_client.py

@@ -65,6 +65,7 @@
     ResponseAwareAuthenticator,
     TokenRotatingAuthenticator,
 )
+from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter
 from airbyte_cdk.sources.utils.types import JsonType
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
 from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH
@@ -160,6 +161,8 @@ def __init__(
         error_message_parser: Optional[ErrorMessageParser] = None,
         disable_retries: bool = False,
         message_repository: Optional[MessageRepository] = None,
+        request_timeout: Optional[Union[float, Tuple[Optional[float], Optional[float]]]] = None,
+        use_tcp_keepalive: bool = False,
     ):
         self._name = name
         self._api_budget: APIBudget = api_budget or APIBudget(policies=[])
@@ -168,12 +171,21 @@ def __init__(
         else:
             self._use_cache = use_cache
             self._session = self._request_session()
-            self._session.mount(
-                "https://",
-                requests.adapters.HTTPAdapter(
-                    pool_connections=MAX_CONNECTION_POOL_SIZE, pool_maxsize=MAX_CONNECTION_POOL_SIZE
-                ),
-            )
+            if use_tcp_keepalive:
+                tcp_keepalive_adapter = TcpKeepaliveHTTPAdapter(
+                    pool_connections=MAX_CONNECTION_POOL_SIZE,
+                    pool_maxsize=MAX_CONNECTION_POOL_SIZE,
+                )
+                self._session.mount("https://", tcp_keepalive_adapter)
+                self._session.mount("http://", tcp_keepalive_adapter)
+            else:
+                self._session.mount(
+                    "https://",
+                    requests.adapters.HTTPAdapter(
+                        pool_connections=MAX_CONNECTION_POOL_SIZE,
+                        pool_maxsize=MAX_CONNECTION_POOL_SIZE,
+                    ),
+                )
         if isinstance(authenticator, AuthBase):
             self._session.auth = authenticator
         self._logger = logger
@@ -191,6 +203,7 @@ def __init__(
         self._disable_retries = disable_retries
         self._message_repository = message_repository
         self._authenticator_update_failed = False
+        self._request_timeout = request_timeout
 
     @property
     def cache_filename(self) -> str:
@@ -453,6 +466,9 @@ def _send(
         response: Optional[requests.Response] = None
         exc: Optional[requests.RequestException] = None
 
+        if self._request_timeout is not None and "timeout" not in request_kwargs:
+            request_kwargs = {**request_kwargs, "timeout": self._request_timeout}
+
         try:
             response = self._session.send(request, **request_kwargs)
         except requests.RequestException as e:

airbyte_cdk/sources/streams/http/tcp_keepalive.py

@@ -0,0 +1,89 @@
+#
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+#
+
+import functools
+import logging
+import socket
+from typing import List, Tuple, cast
+
+import requests
+import urllib3.connection
+
+TCP_KEEPALIVE_IDLE_SECONDS = 60
+TCP_KEEPALIVE_INTERVAL_SECONDS = 10
+TCP_KEEPALIVE_PROBE_COUNT = 6
+
+logger = logging.getLogger("airbyte")
+
+
+@functools.lru_cache(maxsize=None)
+def tcp_keepalive_socket_options() -> Tuple[Tuple[int, int, int], ...]:
+    """Socket options enabling TCP keepalive, probed for what this OS supports.
+
+    Returns urllib3's default options plus SO_KEEPALIVE and the platform's idle /
+    interval / probe-count tuning options (TCP_KEEPIDLE on Linux/Windows,
+    TCP_KEEPALIVE on macOS). Each option is tried once on a throwaway socket and
+    dropped if the OS rejects it, because urllib3 applies these at connect time
+    and an unsupported option would fail every connection. Never raises.
+    """
+    base_options = cast(
+        List[Tuple[int, int, int]],
+        list(urllib3.connection.HTTPConnection.default_socket_options),
+    )
+
+    idle_option = getattr(socket, "TCP_KEEPIDLE", None) or getattr(socket, "TCP_KEEPALIVE", None)
+    candidates: List[Tuple[int, int, int]] = [
+        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
+    ]
+    if idle_option is not None:
+        candidates.append((socket.IPPROTO_TCP, idle_option, TCP_KEEPALIVE_IDLE_SECONDS))
+    for name, value in (
+        ("TCP_KEEPINTVL", TCP_KEEPALIVE_INTERVAL_SECONDS),
+        ("TCP_KEEPCNT", TCP_KEEPALIVE_PROBE_COUNT),
+    ):
+        option = getattr(socket, name, None)
+        if option is not None:
+            candidates.append((socket.IPPROTO_TCP, option, value))
+
+    try:
+        probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    except OSError as error:
+        _warn_keepalive_unavailable(error)
+        return tuple(base_options)
+    try:
+        supported = base_options
+        for level, optname, value in candidates:
+            try:
+                probe.setsockopt(level, optname, value)
+            except OSError as error:
+                if (level, optname) == (socket.SOL_SOCKET, socket.SO_KEEPALIVE):
+                    # Without SO_KEEPALIVE the tuning options are useless.
+                    _warn_keepalive_unavailable(error)
+                    return tuple(base_options)
+                continue
+            supported.append((level, optname, value))
+        return tuple(supported)
+    finally:
+        probe.close()
+
+
+def _warn_keepalive_unavailable(error: OSError) -> None:
+    # Logged once per process because tcp_keepalive_socket_options() is cached.
+    logger.warning(
+        "TCP keepalive is unavailable on this platform, so connections use plain sockets: %s", error
+    )
+
+
+class TcpKeepaliveHTTPAdapter(requests.adapters.HTTPAdapter):
+    """HTTPAdapter that enables TCP keepalive on every connection it pools."""
+
+    def init_poolmanager(self, *args, **pool_kwargs):  # type: ignore[no-untyped-def]
+        pool_kwargs.setdefault("socket_options", tcp_keepalive_socket_options())
+        super().init_poolmanager(*args, **pool_kwargs)  # type: ignore[no-untyped-call]
+
+    def proxy_manager_for(self, proxy, **proxy_kwargs):  # type: ignore[no-untyped-def]
+        # requests does not forward init_poolmanager kwargs to proxy managers,
+        # so without this override proxied connections get no keepalive.
+        proxy_kwargs.setdefault("socket_options", tcp_keepalive_socket_options())
+        return super().proxy_manager_for(proxy, **proxy_kwargs)  # type: ignore[no-untyped-call]

pyproject.toml

@@ -58,6 +58,7 @@ requests_cache = "*"
 rich = "*"  # Use whichever version meets other constraints
 rich-click = "^1.8.8"
 typing-extensions = "*"  # Transitive dependency used directly in code
+urllib3 = "*"  # Transitive dependency used directly in code
 wcmatch = "10.0"
 # Extras depedencies
 avro = { version = ">=1.11.2,<1.13.0", optional = true }  # TODO: Move into dev dependencies if only used in tests

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -274,6 +274,7 @@
 from airbyte_cdk.sources.streams.http.requests_native_auth.oauth import (
     SingleUseRefreshTokenOauth2Authenticator,
 )
+from airbyte_cdk.sources.streams.http.tcp_keepalive import TcpKeepaliveHTTPAdapter
 from airbyte_cdk.sources.types import StreamSlice
 from airbyte_cdk.utils import AirbyteTracedException
 from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
@@ -10096,3 +10097,88 @@ def test_create_file_uploader_with_a_combined_download_target_extractor():
     )
 
     assert isinstance(file_uploader.download_target_extractor, CombinedExtractor)
+
+
+@pytest.mark.parametrize(
+    "manifest_fields, expected_timeout",
+    [
+        pytest.param(
+            {"connect_timeout_in_seconds": 30, "read_timeout_in_seconds": 300},
+            (30.0, 300.0),
+            id="both_timeouts",
+        ),
+        pytest.param(
+            {"read_timeout_in_seconds": 300},
+            (None, 300.0),
+            id="read_only",
+        ),
+        pytest.param(
+            {"connect_timeout_in_seconds": 30},
+            (30.0, None),
+            id="connect_only",
+        ),
+        pytest.param({}, None, id="no_timeouts"),
+    ],
+)
+def test_create_http_requester_with_request_timeouts(manifest_fields, expected_timeout):
+    requester = factory.create_component(
+        model_type=HttpRequesterModel,
+        component_definition={
+            "type": "HttpRequester",
+            "url_base": "https://api.test.com",
+            "path": "/data",
+            "http_method": "GET",
+            **manifest_fields,
+        },
+        config=input_config,
+        name="name",
+        decoder=None,
+    )
+
+    assert requester._http_client._request_timeout == expected_timeout
+    if expected_timeout is None:
+        assert (
+            type(requester._http_client._session.adapters["https://"])
+            is requests.adapters.HTTPAdapter
+        )
+        assert (
+            type(requester._http_client._session.adapters["http://"])
+            is requests.adapters.HTTPAdapter
+        )
+
+
+def test_create_http_requester_with_tcp_keepalive():
+    requester = factory.create_component(
+        model_type=HttpRequesterModel,
+        component_definition={
+            "type": "HttpRequester",
+            "url_base": "https://api.test.com",
+            "path": "/data",
+            "http_method": "GET",
+            "use_tcp_keepalive": True,
+        },
+        config=input_config,
+        name="name",
+        decoder=None,
+    )
+
+    assert isinstance(requester._http_client._session.adapters["https://"], TcpKeepaliveHTTPAdapter)
+    assert isinstance(requester._http_client._session.adapters["http://"], TcpKeepaliveHTTPAdapter)
+
+
+@pytest.mark.parametrize(
+    "manifest_fields",
+    [
+        pytest.param({"read_timeout_in_seconds": 0}, id="read_zero"),
+        pytest.param({"connect_timeout_in_seconds": -1}, id="connect_negative"),
+        pytest.param({"read_timeout_in_seconds": -5}, id="read_negative"),
+    ],
+)
+def test_http_requester_rejects_non_positive_timeouts(manifest_fields):
+    with pytest.raises(ValidationError):
+        HttpRequesterModel(
+            type="HttpRequester",
+            url_base="https://api.test.com",
+            path="/data",
+            **manifest_fields,
+        )