CDK: updated error message for missing streams (#36833)

airbyte-cdk/python/airbyte_cdk/sources/abstract_source.py

@@ -113,9 +113,16 @@ def read(
                 if not stream_instance:
                     if not self.raise_exception_on_missing_stream:
                         continue
-                    raise KeyError(
-                        f"The stream {configured_stream.stream.name} no longer exists in the configuration. "
-                        f"Refresh the schema in replication settings and remove this stream from future sync attempts."
+
+                    error_message = (
+                        f"The stream '{configured_stream.stream.name}' in your connection configuration was not found in the source. "
+                        f"Refresh the schema in your replication settings and remove this stream from future sync attempts."
+                    )
+
+                    raise AirbyteTracedException(
+                        message="A stream listed in your configuration was not found in the source. Please check the logs for more details.",
+                        internal_message=error_message,
+                        failure_type=FailureType.config_error,
                     )
 
                 try:

airbyte-cdk/python/airbyte_cdk/sources/concurrent_source/concurrent_source_adapter.py

@@ -5,12 +5,13 @@
 from abc import ABC
 from typing import Any, Iterator, List, Mapping, MutableMapping, Optional, Union
 
-from airbyte_cdk.models import AirbyteMessage, AirbyteStateMessage, ConfiguredAirbyteCatalog
+from airbyte_cdk.models import AirbyteMessage, AirbyteStateMessage, ConfiguredAirbyteCatalog, FailureType
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.concurrent_source.concurrent_source import ConcurrentSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 from airbyte_cdk.sources.streams.concurrent.abstract_stream_facade import AbstractStreamFacade
+from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 
 class ConcurrentSourceAdapter(AbstractSource, ABC):
@@ -54,10 +55,18 @@ def _select_abstract_streams(self, config: Mapping[str, Any], configured_catalog
             if not stream_instance:
                 if not self.raise_exception_on_missing_stream:
                     continue
-                raise KeyError(
-                    f"The stream {configured_stream.stream.name} no longer exists in the configuration. "
-                    f"Refresh the schema in replication settings and remove this stream from future sync attempts."
+
+                error_message = (
+                    f"The stream '{configured_stream.stream.name}' in your connection configuration was not found in the source. "
+                    f"Refresh the schema in your replication settings and remove this stream from future sync attempts."
+                )
+
+                raise AirbyteTracedException(
+                    message="A stream listed in your configuration was not found in the source. Please check the logs for more details.",
+                    internal_message=error_message,
+                    failure_type=FailureType.config_error,
                 )
+
             if isinstance(stream_instance, AbstractStreamFacade):
                 abstract_streams.append(stream_instance.get_underlying_stream())
         return abstract_streams