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