From fb799dd593e5d41300e93fdd090817f0324785e1 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 24 Mar 2026 12:01:41 +0000 Subject: [PATCH] fix(cdk): reclassify async job failure from config_error to system_error and add user-facing message - Change FailureType from config_error to system_error in _process_partitions_with_errors and the final aggregated exception in create_and_get_completed_partitions - Add user-facing message field to both AirbyteTracedException instances - Update test assertion to expect system_error instead of config_error The previous config_error classification incorrectly prevented automatic retries and misled oncall into thinking these were user configuration issues. These are job execution failures (e.g., expired download URLs, HTTP 403) that are not caused by user configuration. Co-Authored-By: bot_apk --- .../sources/declarative/async_job/job_orchestrator.py | 7 ++++--- .../sources/declarative/async_job/test_job_orchestrator.py | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) 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(