diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py
index 8f70b6e352..c1d0eabd67 100644
--- a/airbyte_cdk/sources/streams/http/http_client.py
+++ b/airbyte_cdk/sources/streams/http/http_client.py
@@ -298,7 +298,22 @@ def _send_with_retry(
 
             return response
         except BaseBackoffException as e:
-            self._logger.error(f"Retries exhausted with backoff exception.", exc_info=True)
+            self._logger.error("Retries exhausted with backoff exception.", exc_info=True)
+
+            is_rate_limited = (
+                isinstance(e.response, requests.Response)
+                and e.response.status_code == requests.codes.too_many_requests
+            )
+
+            if is_rate_limited:
+                raise AirbyteTracedException(
+                    internal_message=f"Rate limit retry budget exhausted. Last exception: {e}",
+                    message="API rate limit exceeded.",
+                    failure_type=FailureType.transient_error,
+                    exception=e,
+                    stream_descriptor=StreamDescriptor(name=self._name),
+                )
+
             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}",
diff --git a/unit_tests/sources/streams/http/test_http.py b/unit_tests/sources/streams/http/test_http.py
index b67d53613f..7f4f9ac412 100644
--- a/unit_tests/sources/streams/http/test_http.py
+++ b/unit_tests/sources/streams/http/test_http.py
@@ -222,7 +222,8 @@ def get_error_handler(self) -> Optional[ErrorHandler]:
     send_mock = mocker.patch.object(requests.Session, "send", return_value=req)
 
     with pytest.raises(
-        AirbyteTracedException, match="Exception: HTTP Status Code: 429. Error: Too many requests."
+        AirbyteTracedException,
+        match="API rate limit exceeded.",
     ):
         list(stream.read_records(SyncMode.full_refresh))
     if retries <= 0:
@@ -316,7 +317,7 @@ def test_raise_on_http_errors_off_429(mocker):
     mocker.patch.object(requests.Session, "send", return_value=req)
     with pytest.raises(
         AirbyteTracedException,
-        match="Exhausted available request attempts. Please see logs for more details. Exception: HTTP Status Code: 429. Error: Too many requests.",
+        match="API rate limit exceeded.",
     ):
         stream.exit_on_rate_limit = True
         list(stream.read_records(SyncMode.full_refresh))