diff --git a/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py b/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py index 0e957011cd..21bb3b071c 100644 --- a/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py +++ b/airbyte_cdk/sources/declarative/async_job/job_orchestrator.py @@ -434,8 +434,9 @@ def _process_partitions_with_errors(self, partition: AsyncPartition) -> None: status_by_job_id = {job.api_job_id(): job.status() for job in partition.jobs} self._non_breaking_exceptions.append( AirbyteTracedException( + message="Async job failed after exhausting all retry attempts.", internal_message=f"At least one job could not be completed for slice {partition.stream_slice}. Job statuses were: {status_by_job_id}. See warning logs for more information.", - failure_type=FailureType.config_error, + failure_type=FailureType.system_error, ) ) @@ -481,14 +482,14 @@ def create_and_get_completed_partitions(self) -> Iterable[AsyncPartition]: # We emitted traced message but we didn't break on non_breaking_exception. We still need to raise an exception so that the # call of `create_and_get_completed_partitions` knows that there was an issue with some partitions and the sync is incomplete. raise AirbyteTracedException( - message=None, + message="One or more async jobs failed after exhausting all retry attempts.", internal_message="\n".join( [ filter_secrets(exception.__repr__()) for exception in self._non_breaking_exceptions ] ), - failure_type=FailureType.config_error, + failure_type=FailureType.system_error, ) def _handle_non_breaking_error(self, exception: Exception) -> None: diff --git a/unit_tests/sources/declarative/async_job/test_job_orchestrator.py b/unit_tests/sources/declarative/async_job/test_job_orchestrator.py index 727f517c43..7fe9bcdf01 100644 --- a/unit_tests/sources/declarative/async_job/test_job_orchestrator.py +++ b/unit_tests/sources/declarative/async_job/test_job_orchestrator.py @@ -296,7 +296,7 @@ def test_given_exception_when_start_job_and_skip_this_exception( assert len(partitions) == 1 # only _job_for_another_slice has succeeded assert self._message_repository.emit_message.call_count == 3 # one for each traced message - assert exception.failure_type == FailureType.config_error # type: ignore # exception should be of type AirbyteTracedException + assert exception.failure_type == FailureType.system_error # type: ignore # exception should be of type AirbyteTracedException @mock.patch(sleep_mock_target) def test_given_jobs_failed_more_than_max_attempts_when_create_and_get_completed_partitions_then_free_job_budget(