fix: update rate limit messages to reference num_workers and keep HTTP status code

Co-Authored-By: sophie.cui@airbyte.io <sophie.cui@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/streams/http/error_handlers/default_error_mapping.py

@@ -61,7 +61,7 @@
     429: ErrorResolution(
         response_action=ResponseAction.RATE_LIMITED,
         failure_type=FailureType.transient_error,
-        error_message="Rate limit exceeded. Try decreasing concurrency or the number of workers to stay within API rate limits.",
+        error_message="Rate limit exceeded (HTTP status code 429). Try decreasing the number of workers to stay within API rate limits.",
     ),
     500: ErrorResolution(
         response_action=ResponseAction.RETRY,

airbyte_cdk/sources/streams/http/http_client.py

@@ -297,7 +297,7 @@ def _send_with_retry(
                 self._logger.error("Rate limit retries exhausted.", exc_info=True)
                 raise AirbyteTracedException(
                     internal_message=f"Rate limit retries exhausted. Exception: {e}",
-                    message="Rate limit exceeded and retries exhausted. Try decreasing concurrency or the number of workers to stay within API rate limits.",
+                    message="Rate limit exceeded and retries exhausted. Try decreasing the number of workers to stay within API rate limits.",
                     failure_type=e.failure_type or FailureType.transient_error,
                     exception=e,
                     stream_descriptor=StreamDescriptor(name=self._name),

airbyte_cdk/sources/streams/http/rate_limiting.py

@@ -147,7 +147,7 @@ def log_retry_attempt(details: Mapping[str, Any]) -> None:
             )
         logger.info(
             f"Rate limit hit after {details['tries']} tries. Waiting {details['wait']} seconds then retrying. "
-            f"Try decreasing concurrency or the number of workers to stay within API rate limits."
+            f"Try decreasing the number of workers to stay within API rate limits."
         )
 
     return backoff.on_exception(  # type: ignore # Decorator function returns a function with a different signature than the input function, so mypy can't infer the type of the returned function

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

@@ -59,7 +59,7 @@
             ErrorResolution(
                 response_action=ResponseAction.RETRY,
                 failure_type=FailureType.transient_error,
-                error_message="Rate limit exceeded. Try decreasing concurrency or the number of workers to stay within API rate limits.",
+                error_message="Rate limit exceeded (HTTP status code 429). Try decreasing the number of workers to stay within API rate limits.",
             ),
             id="test_http_code_matches_retry_action",
         ),

unit_tests/sources/streams/http/test_http.py

@@ -223,7 +223,7 @@ def get_error_handler(self) -> Optional[ErrorHandler]:
 
     with pytest.raises(
         AirbyteTracedException,
-        match="Rate limit exceeded. Try decreasing concurrency or the number of workers to stay within API rate limits.",
+        match="Rate limit exceeded \\(HTTP status code 429\\). Try decreasing the number of workers to stay within API rate limits.",
     ):
         list(stream.read_records(SyncMode.full_refresh))
     if retries <= 0:
@@ -317,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="Rate limit exceeded. Try decreasing concurrency or the number of workers to stay within API rate limits.",
+        match="Rate limit exceeded \\(HTTP status code 429\\). Try decreasing the number of workers to stay within API rate limits.",
     ):
         stream.exit_on_rate_limit = True
         list(stream.read_records(SyncMode.full_refresh))