REST: Add retry and timeout configuration for REST catalog

mkdocs/docs/configuration.md

@@ -413,6 +413,35 @@ catalog:
 
 When server-side planning returns `storage-credentials` on a completed plan, PyIceberg applies them to the scan-scoped FileIO (layered on top of the existing table/load-time IO properties) so planned data and delete files can be read using the creds vended by the server.
 
+#### Retry and timeout
+
+The REST Catalog uses `requests` with no retries and no timeout by default, so transient
+5xx / network failures bubble up immediately and slow servers can hang the client indefinitely.
+Set the `rest.client.*` catalog properties to opt in to a request timeout and a retry policy.
+The property names mirror the Java REST client so a single catalog configuration can serve both.
+
+```yaml
+catalog:
+  default:
+    uri: http://rest-catalog/ws/
+    rest.client.connection-timeout-ms: 5000  # milliseconds, time allowed to establish a connection
+    rest.client.socket-timeout-ms: 60000     # milliseconds, time allowed between bytes once connected
+    rest.client.max-retries: 5               # number of retry attempts on transient failures
+    rest.client.retry-backoff-factor: 1.0    # exponential backoff between retries
+```
+
+| Key                                     | Example  | Description                                                                                                        |
+| --------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------ |
+| rest.client.connection-timeout-ms       | 5000     | Time to establish a connection, in milliseconds. Must be a positive number.                                        |
+| rest.client.socket-timeout-ms           | 60000    | Time allowed between bytes once a connection is established, in milliseconds. Must be a positive number.            |
+| rest.client.max-retries                 | 5        | Number of retry attempts for transient failures. Must be non-negative.                                             |
+| rest.client.retry-backoff-factor        | 1.0      | Backoff factor between retry attempts. Must be non-negative. See [`urllib3` Retry docs](https://urllib3.readthedocs.io/en/latest/reference/urllib3.util.html#urllib3.util.Retry) for the formula. |
+
+`requests` cannot split the connection and socket timeouts, so the two values are summed and applied
+as a single request timeout (floored to whole seconds). Retries are applied to idempotent methods
+only (`GET`, `HEAD`, `OPTIONS`, `PUT`, `DELETE`) and to the transient HTTP status codes `429`, `500`,
+`502`, `503`, `504`. Other failures are not retried.
+
 #### Headers in REST Catalog
 
 To configure custom headers in REST Catalog, include them in the catalog properties with `header.<Header-Name>`. This

pyiceberg/catalog/rest/__init__.py

@@ -19,6 +19,7 @@
 import logging
 import time
 from collections import deque
+from collections.abc import Mapping
 from enum import Enum
 from typing import (
     TYPE_CHECKING,
@@ -27,9 +28,11 @@
 from urllib.parse import quote, unquote
 
 from pydantic import ConfigDict, Field, TypeAdapter, field_validator
-from requests import HTTPError, Session
+from requests import HTTPError, PreparedRequest, Response, Session
+from requests.adapters import DEFAULT_RETRIES, HTTPAdapter
 from tenacity import RetryCallState, retry, retry_if_exception_type, stop_after_attempt
 from typing_extensions import override
+from urllib3.util.retry import Retry
 
 from pyiceberg import __version__
 from pyiceberg.catalog import BOTOCORE_SESSION, TOKEN, URI, WAREHOUSE_LOCATION, Catalog, PropertiesUpdateSummary
@@ -94,7 +97,13 @@
 from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel, Identifier, Properties
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
-from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool, property_as_int
+from pyiceberg.utils.properties import (
+    get_first_property_value,
+    get_header_properties,
+    property_as_bool,
+    property_as_float,
+    property_as_int,
+)
 from pyiceberg.view import View
 from pyiceberg.view.metadata import ViewMetadata, ViewVersion
 
@@ -274,6 +283,14 @@ class ScanPlanningMode(Enum):
 SIGV4_SERVICE = "rest.signing-name"
 SIGV4_MAX_RETRIES = "rest.sigv4.max-retries"
 SIGV4_MAX_RETRIES_DEFAULT = 10
+REST_CLIENT_CONNECTION_TIMEOUT_MS = "rest.client.connection-timeout-ms"
+REST_CLIENT_SOCKET_TIMEOUT_MS = "rest.client.socket-timeout-ms"
+REST_CLIENT_MAX_RETRIES = "rest.client.max-retries"
+REST_CLIENT_RETRY_BACKOFF_FACTOR = "rest.client.retry-backoff-factor"
+# Hard-coded internally so users cannot misconfigure the retry policy
+# (e.g. setting raise_on_status=False would swallow 4xx errors silently).
+_CONNECTION_RETRY_STATUS_FORCELIST = (429, 500, 502, 503, 504)
+_CONNECTION_RETRY_ALLOWED_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "PUT", "DELETE"})
 EMPTY_BODY_SHA256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
@@ -442,6 +459,81 @@ class ListViewsResponse(IcebergBaseModel):
 _PLANNING_RESPONSE_ADAPTER = TypeAdapter(PlanningResponse)
 
 
+class _RetryTimeoutHTTPAdapter(HTTPAdapter):
+    """HTTPAdapter that applies a default per-request timeout.
+
+    requests does not provide a way to set a default timeout on a Session;
+    without this adapter, every call would have to thread `timeout=` through.
+    The adapter applies `self._timeout` whenever a per-call timeout is not set.
+    """
+
+    def __init__(self, timeout: float | None = None, max_retries: Retry | int = DEFAULT_RETRIES) -> None:
+        self._timeout = timeout
+        super().__init__(max_retries=max_retries)
+
+    def send(
+        self,
+        request: PreparedRequest,
+        stream: bool = False,
+        timeout: None | float | tuple[float, float] | tuple[float, None] = None,
+        verify: bool | str = True,
+        cert: None | bytes | str | tuple[bytes | str, bytes | str] = None,
+        proxies: Mapping[str, str] | None = None,
+    ) -> Response:
+        if timeout is None:
+            timeout = self._timeout
+        return super().send(request, stream=stream, timeout=timeout, verify=verify, cert=cert, proxies=proxies)
+
+
+def _create_connection_adapter(properties: Properties) -> _RetryTimeoutHTTPAdapter | None:
+    """Build a connection adapter from the optional `rest.client.*` properties.
+
+    Returns None when no connection properties are supplied, leaving the default
+    Session behavior unchanged. Raises ValueError on invalid input.
+    """
+    connection_timeout_ms = property_as_int(properties, REST_CLIENT_CONNECTION_TIMEOUT_MS)
+    if connection_timeout_ms is not None and connection_timeout_ms <= 0:
+        raise ValueError(f"`{REST_CLIENT_CONNECTION_TIMEOUT_MS}` must be a positive number, got: {connection_timeout_ms}")
+
+    socket_timeout_ms = property_as_int(properties, REST_CLIENT_SOCKET_TIMEOUT_MS)
+    if socket_timeout_ms is not None and socket_timeout_ms <= 0:
+        raise ValueError(f"`{REST_CLIENT_SOCKET_TIMEOUT_MS}` must be a positive number, got: {socket_timeout_ms}")
+
+    retries = property_as_int(properties, REST_CLIENT_MAX_RETRIES)
+    if retries is not None and retries < 0:
+        raise ValueError(f"`{REST_CLIENT_MAX_RETRIES}` must be non-negative, got: {retries}")
+
+    backoff_factor = property_as_float(properties, REST_CLIENT_RETRY_BACKOFF_FACTOR)
+    if backoff_factor is not None and backoff_factor < 0:
+        raise ValueError(f"`{REST_CLIENT_RETRY_BACKOFF_FACTOR}` must be non-negative, got: {backoff_factor}")
+
+    if all(value is None for value in (connection_timeout_ms, socket_timeout_ms, retries, backoff_factor)):
+        return None
+
+    # requests uses a single timeout and cannot split connect vs socket, so follow the Java client
+    # and sum the two (milliseconds), flooring to whole seconds.
+    timeout: float | None = None
+    if connection_timeout_ms is not None or socket_timeout_ms is not None:
+        timeout = ((connection_timeout_ms or 0) + (socket_timeout_ms or 0)) // 1000
+
+    return _RetryTimeoutHTTPAdapter(
+        timeout=timeout,
+        max_retries=Retry(
+            # `retries` and `backoff_factor` fall back to a no-op Retry when unset, so a user can
+            # configure only one without having to specify the rest of the policy.
+            total=retries if retries is not None else DEFAULT_RETRIES,
+            backoff_factor=backoff_factor if backoff_factor is not None else 0.0,
+            status_forcelist=list(_CONNECTION_RETRY_STATUS_FORCELIST),
+            allowed_methods=_CONNECTION_RETRY_ALLOWED_METHODS,
+            # Return the final response on retry exhaustion (instead of raising MaxRetryError)
+            # so `_handle_non_200_response` can map the 5xx status to a typed exception
+            # (ServiceUnavailableError, etc.). 4xx codes are not in status_forcelist and are
+            # never retried, so they reach the same mapping unchanged.
+            raise_on_status=False,
+        ),
+    )
+
+
 class RestCatalog(Catalog):
     uri: str
     _session: Session
@@ -468,6 +560,12 @@ def _create_session(self) -> Session:
         """Create a request session with provided catalog configuration."""
         session = Session()
 
+        # Mount the retry/timeout adapter when `connection.*` properties are set.
+        # SigV4's adapter mounted below at `self.uri` is a longer prefix and still wins for that host.
+        if (connection_adapter := _create_connection_adapter(self.properties)) is not None:
+            session.mount("http://", connection_adapter)
+            session.mount("https://", connection_adapter)
+
         # Set HTTP headers
         self._config_headers(session)
 
@@ -980,8 +1078,6 @@ def _init_sigv4(self, session: Session) -> None:
         import boto3
         from botocore.auth import SigV4Auth
         from botocore.awsrequest import AWSRequest
-        from requests import PreparedRequest
-        from requests.adapters import HTTPAdapter
 
         class SigV4Adapter(HTTPAdapter):
             def __init__(self, **properties: str):