From 101a869690ba4090cd24c953333bd5c66954fcfb Mon Sep 17 00:00:00 2001 From: Baz <22987674+bazarnov@users.noreply.github.com> Date: Mon, 5 Oct 2026 22:47:05 +0300 Subject: [PATCH] fix(low-code): report why the API rejected the last window or page When request window splitting or page size reduction runs out, the error that ends the read now carries the error handler's message for the last rejection in its internal message, so a 400 size limit can be told apart from a 504 timeout. - `RequestWindowSplitRequiredException` and `PageSizeReductionRequiredException` keep `error_message`, and their message describes the rejection only. Their "not set up" advice no longer shows up in the traceback of a stream that is set up correctly; the NotSupported exceptions and the factory's config check still carry it. - The depth cap message no longer blames the stream's cursor. Its internal message names the last rejected window. - `PageSizeReducer.reduce` takes the handler message and adds it to the errors it raises once the page size cannot be reduced any further. Fixes #1197 --- .../retrievers/page_size_reducer.py | 18 ++-- .../retrievers/simple_retriever.py | 13 ++- .../http/page_size_reduction_exception.py | 3 +- .../http/request_window_split_exception.py | 7 +- .../retrievers/test_simple_retriever.py | 91 +++++++++++++++++++ .../test_concurrent_declarative_source.py | 6 ++ .../sources/streams/http/test_http_client.py | 10 +- 7 files changed, 126 insertions(+), 22 deletions(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/page_size_reducer.py b/airbyte_cdk/sources/declarative/retrievers/page_size_reducer.py index 496313db2f..28010badfb 100644 --- a/airbyte_cdk/sources/declarative/retrievers/page_size_reducer.py +++ b/airbyte_cdk/sources/declarative/retrievers/page_size_reducer.py @@ -110,10 +110,11 @@ def page_size_override(self) -> Optional[int]: """ return self._current_page_size - def reduce(self) -> None: + def reduce(self, error_message: Optional[str] = None) -> None: """ Shrink the page size used for the next request. Raises once the page size cannot be shrunk any further - so that an endpoint that keeps failing does not loop forever. + so that an endpoint that keeps failing does not loop forever. Those errors include `error_message`, the + error handler's message for the rejected page. """ if self._configured_page_size is None: raise AirbyteTracedException( @@ -139,6 +140,7 @@ def reduce(self) -> None: failure_type=FailureType.config_error, ) + rejection = f"; the API rejected the page: {error_message}" if error_message else "" reduced_page_size = max( self._config.minimum_page_size, int(current_page_size // self._config.reduction_factor), @@ -146,7 +148,7 @@ def reduce(self) -> None: if reduced_page_size >= current_page_size: # Nothing left to give up on the page size. Whether that is the end of the read is # `retries_at_minimum_page_size`'s call, not this branch's: the response may still be transient. - self._retry_at_minimum_page_size(current_page_size) + self._retry_at_minimum_page_size(current_page_size, rejection) return self._attempts += 1 @@ -157,7 +159,7 @@ def reduce(self) -> None: # succeeding restarts this counter on each of them, under either reset policy, and reads to the # end however many pages it has; a partition where nothing gets through burns the budget here. raise AirbyteTracedException( - internal_message=f"Stream {self._stream_name} reduced its page size {self._attempts - 1} times in a row without a single page succeeding, which is the configured maximum of {self._config.max_attempts} ({self._total_reductions - 1} reductions so far while reading this partition)", + internal_message=f"Stream {self._stream_name} reduced its page size {self._attempts - 1} times in a row without a single page succeeding, which is the configured maximum of {self._config.max_attempts} ({self._total_reductions - 1} reductions so far while reading this partition){rejection}", # `transient_error`, so the only remediation is the connector's own, if it defined one. message=self._with_failure_message( f"The source keeps rejecting pages of stream {self._stream_name} at every page size the " @@ -174,7 +176,7 @@ def reduce(self) -> None: self._current_page_size = reduced_page_size self._sleep(backoff) - def _retry_at_minimum_page_size(self, current_page_size: int) -> None: + def _retry_at_minimum_page_size(self, current_page_size: int, rejection: str) -> None: """ Handle a `REDUCE_PAGE_SIZE` response that arrives when the page size is already as small as the connector is allowed to request. @@ -215,7 +217,8 @@ def _retry_at_minimum_page_size(self, current_page_size: int) -> None: f" ({self._retries_at_minimum_page_size} retries at that size were spent first)" if self._retries_at_minimum_page_size else "" - ), + ) + + rejection, message=f"The page size of stream {self._stream_name} ({current_page_size}) is already at or below " f"the configured minimum of {self._config.minimum_page_size}, so the connector cannot reduce it. " f"Raise the page size of the stream, or lower `minimum_page_size`.", @@ -228,7 +231,8 @@ def _retry_at_minimum_page_size(self, current_page_size: int) -> None: f", after {self._retries_at_minimum_page_size} retries at that size" if self._retries_at_minimum_page_size else "" - ), + ) + + rejection, # `transient_error`, so the only remediation is the connector's own, if it defined one. message=self._with_failure_message( f"The source keeps rejecting pages of stream {self._stream_name} at the smallest page size " diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 26f0d616c3..19b9cce00b 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -492,7 +492,7 @@ def _read_pages( yield current_record except PaginationResetRequiredException: reset_pagination = True - except PageSizeReductionRequiredException: + except PageSizeReductionRequiredException as exception: if page_size_reducer is None: # The action can be attached to a requester we cannot validate at config time, such as one # built by a custom error handler or a CustomRequester. Re-raise as the misconfiguration it @@ -511,7 +511,7 @@ def _read_pages( ) # Raises once the page size cannot be reduced any further and the retries allowed at that # floor are spent, which is what stops the loop when the API keeps failing. - page_size_reducer.reduce() + page_size_reducer.reduce(exception.error_message) reduce_page_size = True else: if page_size_reducer: @@ -640,6 +640,7 @@ def _read_records_or_split_request_window( f"it was rejected; splitting and re-reading the window may re-emit them." ) + rejection = f": {exception.error_message}" if exception.error_message else "" if depth >= _MAX_REQUEST_WINDOW_SPLIT_DEPTH: # Logged separately from the exception below so it stays visible even if the trace message's # internal_message isn't surfaced by whatever catches it. @@ -648,13 +649,11 @@ def _read_records_or_split_request_window( f"({_MAX_REQUEST_WINDOW_SPLIT_DEPTH}) while splitting {original_slice}." ) raise AirbyteTracedException( - internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}", + internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}; the API still rejected {stream_slice}{rejection}", # `transient_error`, so the only remediation is the connector's own, if it defined one. message=self._with_request_window_failure_message( f"Stream {self.name} could not split its request window to a size the API accepts " - f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits. This usually means the stream's " - f"cursor is not actually shrinking the window on each split; if it uses a custom " - f"cursor, check its `split_request_window` implementation." + f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits." ), failure_type=FailureType.transient_error, ) from exception @@ -669,7 +668,7 @@ def _read_records_or_split_request_window( else "" ) raise AirbyteTracedException( - internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further", + internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further; the API rejected it{rejection}", # `transient_error` regardless of how the triggering response was classified: exhaustion is # never the user's fault, matching `PageSizeReducer`'s equivalent exhaustion branch. message=self._with_request_window_failure_message( diff --git a/airbyte_cdk/sources/streams/http/page_size_reduction_exception.py b/airbyte_cdk/sources/streams/http/page_size_reduction_exception.py index 451149eecd..563cff2120 100644 --- a/airbyte_cdk/sources/streams/http/page_size_reduction_exception.py +++ b/airbyte_cdk/sources/streams/http/page_size_reduction_exception.py @@ -34,9 +34,10 @@ def __init__( detail = f": {error_message}" if error_message else "" super().__init__( internal_message=f"An error handler{stream} resolved to REDUCE_PAGE_SIZE{detail}", - message=f"The API rejected a page{stream} and asked the connector for a smaller one. If this message ends a sync, the stream is not set up to request a smaller page: add `page_size_reduction` to its retriever, or remove the REDUCE_PAGE_SIZE action from its error handler.", + message=f"The API rejected a page{stream} and asked the connector for a smaller one{detail or '.'}", failure_type=FailureType.config_error, ) + self.error_message = error_message class PageSizeReductionNotSupportedException(AirbyteTracedException): diff --git a/airbyte_cdk/sources/streams/http/request_window_split_exception.py b/airbyte_cdk/sources/streams/http/request_window_split_exception.py index 0ee937fa00..a3ae6ec700 100644 --- a/airbyte_cdk/sources/streams/http/request_window_split_exception.py +++ b/airbyte_cdk/sources/streams/http/request_window_split_exception.py @@ -20,6 +20,10 @@ class RequestWindowSplitRequiredException(AirbyteTracedException): failure the built-in `HttpResponseFilter`/`ResponseAction.SPLIT_REQUEST_WINDOW` path cannot see, without the CDK needing to know about that connector's specific exception types. + It is raised on every split, including the ones a correctly configured stream is expected to make, so the + message describes the rejection and nothing else: it shows up in the traceback of any later failure of the + same read. A split `SimpleRetriever` cannot honor raises `RequestWindowSplitNotSupportedException` instead. + The classifying error's `error_message` is preserved on this exception so it can be surfaced if the window is never split any further. `failure_type` is not: exhausting every window size the API will accept is never the user's fault, so `SimpleRetriever` always reports that as `transient_error` regardless of how the @@ -38,9 +42,10 @@ def __init__( detail = f": {error_message}" if error_message else "" super().__init__( internal_message=f"A request window split was requested{stream}{detail}", - message=f"The API rejected the current request window{stream} and requires a smaller one. If this message ends a sync, the stream is not set up to split its window: add `request_window_splitting` to its retriever, or remove the `SPLIT_REQUEST_WINDOW` action from its error handler.", + message=f"The API rejected the current request window{stream} and requires a smaller one{detail or '.'}", failure_type=failure_type or FailureType.config_error, ) + self.error_message = error_message class RequestWindowSplitNotSupportedException(AirbyteTracedException): diff --git a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py index 4b87f0fed5..c242b9d0b9 100644 --- a/unit_tests/sources/declarative/retrievers/test_simple_retriever.py +++ b/unit_tests/sources/declarative/retrievers/test_simple_retriever.py @@ -1725,6 +1725,39 @@ def test_given_reductions_exhausted_when_read_records_then_raise_transient_error assert requester.send_request.call_count == 3 +@pytest.mark.parametrize( + "page_size_reduction", + [ + pytest.param(PageSizeReduction(max_attempts=2), id="max_attempts_reached"), + pytest.param(PageSizeReduction(minimum_page_size=50), id="minimum_page_size_reached"), + pytest.param(PageSizeReduction(minimum_page_size=100), id="page_size_starts_at_minimum"), + ], +) +def test_given_reductions_exhausted_when_read_records_then_report_the_api_rejection( + page_size_reduction, +): + """ + The error handler's message is the only record of why the API rejected the page, so the error that ends + the read carries it, and its traceback does not claim the stream is not set up to reduce its page size. + """ + requester = Mock(spec=Requester) + requester.send_request.side_effect = PageSizeReductionRequiredException( + stream_name=A_STREAM_NAME, error_message="Query cost 9000 exceeds 5000." + ) + paginator = _mock_paginator() + paginator.get_page_size.return_value = 100 + paginator.get_initial_token.return_value = None + retriever = _page_size_reduction_retriever( + requester, paginator, Mock(spec=HttpSelector), page_size_reduction + ) + + with pytest.raises(AirbyteTracedException) as exception: + list(retriever.read_records(A_RECORD_SCHEMA, A_STREAM_SLICE)) + + assert "Query cost 9000 exceeds 5000." in exception.value.internal_message + assert "not set up" not in exception.value.as_airbyte_message().trace.error.stack_trace + + def test_given_no_page_size_reduction_when_reduce_page_size_required_then_raise_config_error(): requester = Mock(spec=Requester) requester.send_request.side_effect = PageSizeReductionRequiredException() @@ -2647,6 +2680,64 @@ def test_given_max_split_depth_exceeded_when_read_records_then_raise_terminal_er assert request_window_splitter.call_count == _MAX_REQUEST_WINDOW_SPLIT_DEPTH +def _halve_window( + stream_slice: StreamSlice, min_split_window: Optional[timedelta] +) -> List[StreamSlice]: + start, end = stream_slice.cursor_slice["start"], stream_slice.cursor_slice["end"] + middle = (start + end) // 2 + return [ + StreamSlice(partition={}, cursor_slice={"start": start, "end": middle}), + StreamSlice(partition={}, cursor_slice={"start": middle + 1, "end": end}), + ] + + +@pytest.mark.parametrize( + "request_window_splitter, last_window", + [ + pytest.param( + lambda stream_slice, min_split_window: None, + {"start": 0, "end": 4095}, + id="window_cannot_be_split_further", + ), + pytest.param(_halve_window, {"start": 0, "end": 3}, id="max_split_depth_reached"), + ], +) +def test_given_window_still_rejected_when_splitting_is_exhausted_then_report_the_api_rejection( + request_window_splitter, last_window +): + """ + The error handler's message is the only record of why the API rejected the last window, so the error that + ends the read carries it along with that window. A cursor that halved the window on every split is not + blamed for reaching the depth cap, and the traceback does not claim the stream is not set up to split. + """ + + def reject_window(**kwargs: Any) -> None: + raise RequestWindowSplitRequiredException( + stream_name=A_STREAM_NAME, + error_message="HTTP Status Code: 504. Error: Gateway timeout.", + ) + + requester = Mock(spec=Requester) + requester.send_request.side_effect = reject_window + paginator = _mock_paginator() + paginator.get_initial_token.return_value = None + retriever = _request_window_splitting_retriever( + requester, paginator, Mock(spec=HttpSelector), request_window_splitter + ) + + with pytest.raises(AirbyteTracedException) as exception: + list( + retriever.read_records( + A_RECORD_SCHEMA, StreamSlice(partition={}, cursor_slice={"start": 0, "end": 4095}) + ) + ) + + assert "HTTP Status Code: 504. Error: Gateway timeout." in exception.value.internal_message + assert str(last_window) in exception.value.internal_message + assert "split_request_window" not in exception.value.message + assert "not set up" not in exception.value.as_airbyte_message().trace.error.stack_trace + + def test_given_partitions_read_concurrently_then_window_reduction_isolates_between_partitions(): """ One retriever instance - and one shared `request_window_splitter` (the stream's cursor) - is used for every partition diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 47b80df66b..2cb63018fd 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -5165,6 +5165,9 @@ def test_given_reductions_exhausted_when_read_then_emit_a_transient_error(): "keeps rejecting pages of stream" in error.message and "records per page" in error.message for error in errors ) + # the filter defines no `error_message`, so the default mapping's text for 502 is the rejection reported + assert "HTTP Status Code: 502" in errors[0].internal_message + assert "not set up" not in errors[0].stack_trace def _request_window_splitting_manifest(): @@ -5507,6 +5510,9 @@ def test_given_split_window_still_rejected_when_read_then_partition_is_not_check ] assert errors[0].failure_type == FailureType.transient_error assert "could not split its request window" in errors[0].internal_message + # the filter defines no `error_message`, so the default mapping's text for 400 is the rejection reported + assert "HTTP Status Code: 400" in errors[0].internal_message + assert "not set up" not in errors[0].stack_trace final_state = get_states_for_stream(stream_name="Test", messages=messages)[ -1 ].stream.stream_state.__dict__ diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index ad3ee5acd1..2190481190 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -1758,9 +1758,8 @@ def test_send_raises_page_size_reduction_required_exception_with_reduce_page_siz # so its message must describe the event rather than accuse the connector of a bug assert "should be reported" not in exception.value.message assert exception.value.message == ( - "The API rejected a page of stream test and asked the connector for a smaller one. If this message " - "ends a sync, the stream is not set up to request a smaller page: add `page_size_reduction` to its " - "retriever, or remove the REDUCE_PAGE_SIZE action from its error handler." + "The API rejected a page of stream test and asked the connector for a smaller one: " + "test reduce page size message" ) # a retriever that cannot re-issue the page never retries it, so a job-level retry cannot help assert exception.value.failure_type == FailureType.config_error @@ -1871,9 +1870,8 @@ def test_send_raises_request_window_splitting_required_exception_with_split_requ assert http_client._session.send.call_count == 1 assert "test reduce request window message" in exception.value.internal_message assert exception.value.message == ( - "The API rejected the current request window of stream test and requires a smaller one. If this " - "message ends a sync, the stream is not set up to split its window: add `request_window_splitting` " - "to its retriever, or remove the `SPLIT_REQUEST_WINDOW` action from its error handler." + "The API rejected the current request window of stream test and requires a smaller one: " + "test reduce request window message" ) # the classifying error's own failure_type is preserved rather than defaulting to config_error assert exception.value.failure_type == FailureType.config_error