From b3ee3c2799c89c25c7b119c39be5989e5bd6646a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 18:42:27 +0000 Subject: [PATCH 1/4] fix(cdk): preserve error message in concurrent source exception handling In ConcurrentReadProcessor.on_exception(), when the inner exception is not an AirbyteTracedException, the code was passing the StreamThreadException wrapper to from_exception() instead of the inner exception. Since from_exception() does not set a user-facing message for non-ATE inputs, this resulted in message=None, triggering the generic fallback: 'Something went wrong in the connector. See the logs for more details.' Fix: pass the inner exception and explicitly provide the message parameter so the actual error text surfaces to users instead of the generic fallback. Co-Authored-By: bot_apk --- .../sources/concurrent_source/concurrent_read_processor.py | 4 +++- .../streams/concurrent/test_concurrent_read_processor.py | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py b/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py index 905999a4d7..9f61f7838e 100644 --- a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py +++ b/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py @@ -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() def _flag_exception(self, stream_name: str, exception: Exception) -> None: diff --git a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py index a681f75ebb..c0b5adedeb 100644 --- a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py +++ b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py @@ -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( From d86b96c6ac1a42e316a7b3f526a52e197438fba3 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 6 Mar 2026 18:58:05 +0000 Subject: [PATCH 2/4] fix: update test assertions to match inner exception stack traces Co-Authored-By: bot_apk --- .../connector_builder/test_connector_builder_handler.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/unit_tests/connector_builder/test_connector_builder_handler.py b/unit_tests/connector_builder/test_connector_builder_handler.py index 6f0394407b..fee9f3e939 100644 --- a/unit_tests/connector_builder/test_connector_builder_handler.py +++ b/unit_tests/connector_builder/test_connector_builder_handler.py @@ -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( From 0081ad574d183640833ecbe5803a4520682a74c0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Mar 2026 19:44:07 +0000 Subject: [PATCH 3/4] fix: use safe templated message and add regression test per review feedback Co-Authored-By: bot_apk --- .../concurrent_read_processor.py | 2 +- .../test_concurrent_read_processor.py | 46 ++++++++++++++++++- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py b/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py index 9f61f7838e..8839018cad 100644 --- a/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py +++ b/airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py @@ -175,7 +175,7 @@ def on_exception(self, exception: StreamThreadException) -> Iterable[AirbyteMess yield AirbyteTracedException.from_exception( exception.exception, stream_descriptor=stream_descriptor, - message=str(exception.exception), + message=f"An unexpected error occurred in stream {exception.stream_name}: {type(exception.exception).__name__}", ).as_airbyte_message() def _flag_exception(self, stream_name: str, exception: Exception) -> None: diff --git a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py index c0b5adedeb..60465d0fea 100644 --- a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py +++ b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py @@ -16,6 +16,7 @@ AirbyteStreamStatus, AirbyteStreamStatusTraceMessage, AirbyteTraceMessage, + FailureType, StreamDescriptor, SyncMode, TraceType, @@ -577,7 +578,7 @@ 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 "RuntimeError" in exception_messages[0].trace.error.stack_trace - assert exception_messages[0].trace.error.message == "Something went wrong" + assert exception_messages[0].trace.error.message == f"An unexpected error occurred in stream {_STREAM_NAME}: RuntimeError" assert list( handler.on_partition_complete_sentinel( @@ -762,6 +763,49 @@ def test_is_done_is_true_if_all_partitions_are_closed_and_no_streams_are_generat assert handler.is_done() + @freezegun.freeze_time("2020-01-01T00:00:00") + def test_on_exception_non_ate_uses_templated_message_with_correct_failure_type(self): + """Regression test: non-ATE exceptions on Path B produce a safe templated message, not the generic fallback.""" + stream_instances_to_read_from = [self._stream, self._another_stream] + + handler = ConcurrentReadProcessor( + stream_instances_to_read_from, + self._partition_enqueuer, + self._thread_pool_manager, + self._logger, + self._slice_logger, + self._message_repository, + self._partition_reader, + ) + + handler.start_next_partition_generator() + handler.on_partition(self._an_open_partition) + list( + handler.on_partition_generation_completed( + PartitionGenerationCompletedSentinel(self._stream) + ) + ) + list( + handler.on_partition_generation_completed( + PartitionGenerationCompletedSentinel(self._another_stream) + ) + ) + + inner_exception = ValueError("some internal detail: SELECT * FROM secrets") + exception = StreamThreadException(inner_exception, _STREAM_NAME) + + exception_messages = list(handler.on_exception(exception)) + assert len(exception_messages) == 1 + + trace_error = exception_messages[0].trace.error + # User-facing message uses safe template (no raw exception text) + assert trace_error.message == f"An unexpected error occurred in stream {_STREAM_NAME}: ValueError" + # Stack trace comes from the inner exception, not the StreamThreadException wrapper + assert "ValueError" in trace_error.stack_trace + assert "StreamThreadException" not in trace_error.stack_trace + # failure_type defaults to system_error for unhandled exceptions + assert trace_error.failure_type == FailureType.system_error + @freezegun.freeze_time("2020-01-01T00:00:00") def test_start_next_partition_generator(self): stream_instances_to_read_from = [self._stream] From 3cdf5577777d101745adb076b540b092dcf490c0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Mar 2026 19:45:34 +0000 Subject: [PATCH 4/4] style: apply ruff format to test assertions Co-Authored-By: bot_apk --- .../concurrent/test_concurrent_read_processor.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py index 60465d0fea..94654713fa 100644 --- a/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py +++ b/unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py @@ -578,7 +578,10 @@ 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 "RuntimeError" in exception_messages[0].trace.error.stack_trace - assert exception_messages[0].trace.error.message == f"An unexpected error occurred in stream {_STREAM_NAME}: RuntimeError" + assert ( + exception_messages[0].trace.error.message + == f"An unexpected error occurred in stream {_STREAM_NAME}: RuntimeError" + ) assert list( handler.on_partition_complete_sentinel( @@ -799,7 +802,10 @@ def test_on_exception_non_ate_uses_templated_message_with_correct_failure_type(s trace_error = exception_messages[0].trace.error # User-facing message uses safe template (no raw exception text) - assert trace_error.message == f"An unexpected error occurred in stream {_STREAM_NAME}: ValueError" + assert ( + trace_error.message + == f"An unexpected error occurred in stream {_STREAM_NAME}: ValueError" + ) # Stack trace comes from the inner exception, not the StreamThreadException wrapper assert "ValueError" in trace_error.stack_trace assert "StreamThreadException" not in trace_error.stack_trace