Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Patrick Nilan (pnilan) merged 1 commit into from
Mar 24, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions airbyte_cdk/sources/declarative/async_job/job_orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
)

Expand Down Expand Up @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion unit_tests/sources/declarative/async_job/test_job_orchestrator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading