fix(cdk): reclassify async job failure from config_error to system_error and add user-facing message (#961)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

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:

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(