feat(rest): add sigv4 retry configuration defaults (#3008)

Author: Mrutunjay Kinagi

pyiceberg/catalog/rest/__init__.py

@@ -77,7 +77,7 @@
 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
+from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool, property_as_int
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -223,6 +223,10 @@ class IdentifierKind(Enum):
 SIGV4 = "rest.sigv4-enabled"
 SIGV4_REGION = "rest.signing-region"
 SIGV4_SERVICE = "rest.signing-name"
+SIGV4_MAX_RETRIES = "rest.sigv4.max-retries"
+SIGV4_DEFAULT_MAX_RETRIES = 10
+SIGV4_RETRY_BACKOFF_FACTOR = 0.5
+SIGV4_RETRY_STATUS_CODES = (429, 500, 502, 503, 504)
 EMPTY_BODY_SHA256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
@@ -680,11 +684,25 @@ def _init_sigv4(self, session: Session) -> None:
         from botocore.awsrequest import AWSRequest
         from requests import PreparedRequest
         from requests.adapters import HTTPAdapter
+        from urllib3.util.retry import Retry
 
         class SigV4Adapter(HTTPAdapter):
             def __init__(self, **properties: str):
-                super().__init__()
                 self._properties = properties
+                max_retries = property_as_int(self._properties, SIGV4_MAX_RETRIES, SIGV4_DEFAULT_MAX_RETRIES)
+                super().__init__(
+                    max_retries=Retry(
+                        total=max_retries,
+                        status=max_retries,
+                        connect=max_retries,
+                        read=max_retries,
+                        # Keep retries conservative for idempotent calls.
+                        allowed_methods=frozenset({"GET", "HEAD", "OPTIONS"}),
+                        status_forcelist=SIGV4_RETRY_STATUS_CODES,
+                        backoff_factor=SIGV4_RETRY_BACKOFF_FACTOR,
+                        respect_retry_after_header=True,
+                    )
+                )
                 self._boto_session = boto3.Session(
                     region_name=get_first_property_value(self._properties, AWS_REGION),
                     botocore_session=self._properties.get(BOTOCORE_SESSION),

tests/catalog/test_rest.py

@@ -33,6 +33,8 @@
     DEFAULT_ENDPOINTS,
     EMPTY_BODY_SHA256,
     OAUTH2_SERVER_URI,
+    SIGV4_DEFAULT_MAX_RETRIES,
+    SIGV4_MAX_RETRIES,
     SNAPSHOT_LOADING_MODE,
     Capability,
     RestCatalog,
@@ -527,6 +529,44 @@ def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None:
     assert prepared.headers.get("x-amz-content-sha256") != EMPTY_BODY_SHA256
 
 
+def test_sigv4_adapter_default_retry_config(rest_mock: Mocker) -> None:
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": TEST_TOKEN,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+        },
+    )
+
+    adapter = catalog._session.adapters[catalog.uri]
+    assert isinstance(adapter, HTTPAdapter)
+    assert adapter.max_retries.total == SIGV4_DEFAULT_MAX_RETRIES
+    assert 429 in adapter.max_retries.status_forcelist
+
+
+def test_sigv4_adapter_override_retry_config(rest_mock: Mocker) -> None:
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": TEST_TOKEN,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+            SIGV4_MAX_RETRIES: "3",
+        },
+    )
+
+    adapter = catalog._session.adapters[catalog.uri]
+    assert isinstance(adapter, HTTPAdapter)
+    assert adapter.max_retries.total == 3
+
+
 def test_list_tables_404(rest_mock: Mocker) -> None:
     namespace = "examples"
     rest_mock.get(