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 <apk@cognition.ai>

Author: devin-ai-integration[bot]

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:

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(