fix(low-code): honor declared failure_type on retry filters and clean up exhaustion message

airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py

@@ -27,7 +27,8 @@ class HttpResponseFilter:
     Filter to select a response based on its HTTP status code, error message or a predicate.
     If a response matches the filter, the response action, failure_type, and error message are returned as an ErrorResolution object.
     For http_codes declared in the filter, the failure_type will default to `system_error`.
-    To override default failure_type use configured failure_type with ResponseAction.FAIL.
+    A configured failure_type overrides the default for the `FAIL`, `RETRY`, `RATE_LIMITED` and
+    `REFRESH_TOKEN_THEN_RETRY` actions. `IGNORE` and `RESET_PAGINATION` keep the default mapping.
 
     Attributes:
         action (Union[ResponseAction, str]): action to execute if a request matches
@@ -95,7 +96,12 @@ def matches(
                 error_message = self._create_error_message(response_or_exception)
             error_message = error_message or default_error_message
 
-            if self.failure_type and filter_action == ResponseAction.FAIL:
+            if self.failure_type and filter_action in {
+                ResponseAction.FAIL,
+                ResponseAction.RETRY,
+                ResponseAction.RATE_LIMITED,
+                ResponseAction.REFRESH_TOKEN_THEN_RETRY,
+            }:
                 failure_type = self.failure_type
             elif default_mapped_error_resolution:
                 failure_type = default_mapped_error_resolution.failure_type

airbyte_cdk/sources/streams/http/http_client.py

@@ -316,7 +316,7 @@ def _send_with_retry(
 
             raise AirbyteTracedException(
                 internal_message=f"Exhausted available request attempts. Exception: {e}",
-                message=f"Exhausted available request attempts. Please see logs for more details. Exception: {e}",
+                message="Available request retry attempts are exhausted.",
                 failure_type=e.failure_type or FailureType.system_error,
                 exception=e,
                 stream_descriptor=StreamDescriptor(name=self._name),

unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py

@@ -48,6 +48,21 @@
             ),
             id="test_http_code_matches_ignore_action",
         ),
+        pytest.param(
+            ResponseAction.IGNORE,
+            FailureType.config_error,
+            {500},
+            "",
+            "",
+            "",
+            {"status_code": 500},
+            ErrorResolution(
+                response_action=ResponseAction.IGNORE,
+                failure_type=FailureType.transient_error,
+                error_message="HTTP Status Code: 500. Error: Internal server error.",
+            ),
+            id="test_http_code_matches_ignore_action_uses_default_failure_type",
+        ),
         pytest.param(
             ResponseAction.RETRY,
             None,
@@ -178,10 +193,10 @@
             {"status_code": 500},
             ErrorResolution(
                 response_action=ResponseAction.RETRY,
-                failure_type=FailureType.transient_error,
+                failure_type=FailureType.config_error,
                 error_message="rate limits",
             ),
-            id="test_http_code_matches_failure_type_config_error_action_retry_uses_default_failure_type",
+            id="test_http_code_matches_failure_type_config_error_action_retry",
         ),
         pytest.param(
             ResponseAction.RATE_LIMITED,

unit_tests/sources/streams/http/test_http_client.py

@@ -11,6 +11,12 @@
 from requests_cache import CachedRequest
 
 from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.requesters.error_handlers import (
+    DefaultErrorHandler as DeclarativeDefaultErrorHandler,
+)
+from airbyte_cdk.sources.declarative.requesters.error_handlers import (
+    HttpResponseFilter,
+)
 from airbyte_cdk.sources.streams.call_rate import CachedLimiterSession, LimiterSession
 from airbyte_cdk.sources.streams.http import HttpClient
 from airbyte_cdk.sources.streams.http.error_handlers import (
@@ -840,6 +846,46 @@ def backoff_time(self, response_or_exception, attempt_count):
     assert e.value.failure_type == expected_failure_type
 
 
+@pytest.mark.usefixtures("mock_sleep")
+def test_send_request_exhaustion_preserves_declared_failure_type_and_hides_retry_details(
+    requests_mock,
+):
+    error_message = "The connector will retry automatically. Please see logs for more details."
+    error_handler = DeclarativeDefaultErrorHandler(
+        config={},
+        parameters={},
+        max_retries=1,
+        response_filters=[
+            HttpResponseFilter(
+                action=ResponseAction.RETRY,
+                failure_type=FailureType.transient_error,
+                http_codes={200},
+                error_message=error_message,
+                config={},
+                parameters={},
+            )
+        ],
+    )
+    http_client = HttpClient(name="test", logger=MagicMock(), error_handler=error_handler)
+    requests_mock.register_uri(
+        "GET",
+        "https://airbyte.io/",
+        status_code=200,
+        json={"code": 50000},
+        headers={},
+    )
+
+    with pytest.raises(AirbyteTracedException) as exception:
+        http_client.send_request(http_method="get", url="https://airbyte.io/", request_kwargs={})
+
+    assert exception.value.failure_type == FailureType.transient_error
+    assert exception.value.message == "Available request retry attempts are exhausted."
+    assert (
+        exception.value.internal_message == "Exhausted available request attempts. Exception: "
+        f"{error_message}"
+    )
+
+
 class MockOAuthAuthenticator:
     def __init__(self):
         self.access_token = "old_token"