Skip to content

fix(cdk): Preserve error message in concurrent source exception handling (AI-Triage PR) #937

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 4 commits into from
Mar 9, 2026
Merged

fix(cdk): Preserve error message in concurrent source exception handling (AI-Triage PR) #937

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b3ee3c2
fix(cdk): preserve error message in concurrent source exception handling
devin-ai-integration[bot] Mar 6, 2026
d86b96c
fix: update test assertions to match inner exception stack traces
devin-ai-integration[bot] Mar 6, 2026
0081ad5
fix: use safe templated message and add regression test per review fe…
devin-ai-integration[bot] Mar 9, 2026
3cdf557
style: apply ruff format to test assertions
devin-ai-integration[bot] Mar 9, 2026
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
4 changes: 3 additions & 1 deletion airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,9 @@ def on_exception(self, exception: StreamThreadException) -> Iterable[AirbyteMess
yield exception.exception.as_airbyte_message(stream_descriptor=stream_descriptor)
else:
yield AirbyteTracedException.from_exception(
exception, stream_descriptor=stream_descriptor
exception.exception,
stream_descriptor=stream_descriptor,
message=str(exception.exception),
).as_airbyte_message()
Comment thread
pnilan marked this conversation as resolved.

def _flag_exception(self, stream_name: str, exception: Exception) -> None:
Expand Down
4 changes: 2 additions & 2 deletions unit_tests/connector_builder/test_connector_builder_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1288,13 +1288,13 @@ def test_handle_read_external_requests(deployment_mode, url_base, expected_error
pytest.param(
"CLOUD",
"https://10.0.27.27/tokens/bearer",
"StreamThreadException",
"Error while refreshing access token",
id="test_cloud_read_with_private_endpoint",
),
pytest.param(
"CLOUD",
"http://unsecured.protocol/tokens/bearer",
"StreamThreadException",
"Invalid Protocol Scheme",
id="test_cloud_read_with_unsecured_endpoint",
),
pytest.param(
Expand Down
3 changes: 2 additions & 1 deletion unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,7 +576,8 @@ def test_on_exception_return_trace_message_and_on_stream_complete_return_stream_

exception_messages = list(handler.on_exception(exception))
assert len(exception_messages) == 1
assert "StreamThreadException" in exception_messages[0].trace.error.stack_trace
assert "RuntimeError" in exception_messages[0].trace.error.stack_trace
assert exception_messages[0].trace.error.message == "Something went wrong"

assert list(
handler.on_partition_complete_sentinel(
Expand Down
Loading