From c09414243dc469cd5c259c518558496dcc089be3 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:13:52 +0000 Subject: [PATCH] fix(cdk): stop labeling every user-defined backoff as a rate limit in retry logs Co-Authored-By: bot_apk --- .../sources/streams/http/rate_limiting.py | 20 +++-- .../streams/http/test_rate_limiting.py | 78 +++++++++++++++++++ 2 files changed, 93 insertions(+), 5 deletions(-) create mode 100644 unit_tests/sources/streams/http/test_rate_limiting.py diff --git a/airbyte_cdk/sources/streams/http/rate_limiting.py b/airbyte_cdk/sources/streams/http/rate_limiting.py index 0137aa5043..49f80b6a86 100644 --- a/airbyte_cdk/sources/streams/http/rate_limiting.py +++ b/airbyte_cdk/sources/streams/http/rate_limiting.py @@ -108,14 +108,24 @@ def sleep_on_ratelimit(details: Mapping[str, Any]) -> None: if isinstance(exc, UserDefinedBackoffException): retry_after = exc.backoff sleep_time = retry_after + 1 # extra second to cover any fractions of second - if exc.response is not None: - logger.info( - f"UserDefinedBackoffException: Rate limit exceeded (HTTP {exc.response.status_code}). Retrying in {sleep_time} seconds." + attempt = details["tries"] + wait = f"{sleep_time:.1f}" + if exc.response is not None and exc.response.status_code == codes.too_many_requests: + message = ( + f"Rate limit exceeded (HTTP 429). Backing off for {wait} seconds " + f"(attempt {attempt}), then retrying." + ) + elif exc.response is not None: + message = ( + f"Backing off for {wait} seconds after HTTP {exc.response.status_code} " + f"response (attempt {attempt}), then retrying." ) else: - logger.info( - f"UserDefinedBackoffException: Rate limit exceeded. Retrying in {sleep_time} seconds." + message = ( + f"Backing off for {wait} seconds after a request error " + f"(attempt {attempt}), then retrying." ) + logger.info(message) time.sleep(sleep_time) def log_give_up(details: Mapping[str, Any]) -> None: diff --git a/unit_tests/sources/streams/http/test_rate_limiting.py b/unit_tests/sources/streams/http/test_rate_limiting.py new file mode 100644 index 0000000000..9572f4acfe --- /dev/null +++ b/unit_tests/sources/streams/http/test_rate_limiting.py @@ -0,0 +1,78 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +import logging +from types import SimpleNamespace + +import pytest +import requests + +from airbyte_cdk.sources.streams.http.exceptions import UserDefinedBackoffException +from airbyte_cdk.sources.streams.http.rate_limiting import user_defined_backoff_handler + + +def _make_request_and_response( + status_code: int, +) -> tuple[requests.PreparedRequest, requests.Response]: + request = requests.Request("GET", "https://example.com/api").prepare() + response = requests.Response() + response.status_code = status_code + return request, response + + +@pytest.mark.parametrize( + ("status_code", "backoff", "request_error", "expected_log", "expect_rate_limit"), + [ + ( + 500, + 2910.8324723243713, + False, + "Backing off for 2911.8 seconds after HTTP 500 response (attempt 1), then retrying.", + False, + ), + ( + 429, + 59, + False, + "Rate limit exceeded (HTTP 429). Backing off for 60.0 seconds (attempt 1), then retrying.", + True, + ), + ( + 500, + 4, + True, + "Backing off for 5.0 seconds after a request error (attempt 1), then retrying.", + False, + ), + ], +) +def test_user_defined_backoff_handler_logs_retry_reason( + mocker, caplog, status_code, backoff, request_error, expected_log, expect_rate_limit +): + request, response = _make_request_and_response(status_code) + response_or_error = requests.exceptions.ReadTimeout("timed out") if request_error else response + sentinel = object() + calls = 0 + + def send_request(): + nonlocal calls + calls += 1 + if calls == 1: + raise UserDefinedBackoffException( + backoff=backoff, request=request, response=response_or_error + ) + return sentinel + + sleep = mocker.patch("airbyte_cdk.sources.streams.http.rate_limiting.time.sleep") + mocker.patch("backoff._sync.time", SimpleNamespace(sleep=lambda _: None)) + wrapped = user_defined_backoff_handler(max_tries=2)(send_request) + + with caplog.at_level(logging.INFO, logger="airbyte"): + result = wrapped() + + messages = [record.getMessage() for record in caplog.records] + assert messages == [expected_log] + assert ("Rate limit" in messages[0]) is expect_rate_limit + assert sleep.call_count == 1 + sleep.assert_called_once_with(backoff + 1) + assert result is sentinel