airbyte_cdk.sources.declarative.concurrent_declarative_source

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2
  3import json
  4import logging
  5import pkgutil
  6from copy import deepcopy
  7from dataclasses import dataclass, field
  8from queue import Queue
  9from types import ModuleType
 10from typing import (
 11    Any,
 12    ClassVar,
 13    Dict,
 14    Iterator,
 15    List,
 16    Mapping,
 17    Optional,
 18    Set,
 19)
 20
 21import orjson
 22import yaml
 23from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor
 24from jsonschema.exceptions import ValidationError
 25from jsonschema.validators import validate
 26
 27from airbyte_cdk.config_observation import create_connector_config_control_message
 28from airbyte_cdk.connector_builder.models import (
 29    LogMessage as ConnectorBuilderLogMessage,
 30)
 31from airbyte_cdk.manifest_migrations.migration_handler import (
 32    ManifestMigrationHandler,
 33)
 34from airbyte_cdk.models import (
 35    AirbyteCatalog,
 36    AirbyteConnectionStatus,
 37    AirbyteMessage,
 38    AirbyteStateMessage,
 39    ConfiguredAirbyteCatalog,
 40    ConnectorSpecification,
 41    FailureType,
 42    Status,
 43)
 44from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer
 45from airbyte_cdk.sources import Source
 46from airbyte_cdk.sources.concurrent_source.concurrent_source import ConcurrentSource
 47from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
 48from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 49from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 50from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
 51from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean
 52from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 53    ConcurrencyLevel as ConcurrencyLevelModel,
 54)
 55from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 56    DeclarativeStream as DeclarativeStreamModel,
 57)
 58from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 59    Spec as SpecModel,
 60)
 61from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 62    StateDelegatingStream as StateDelegatingStreamModel,
 63)
 64from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
 65    get_registered_components_module,
 66)
 67from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import (
 68    ManifestComponentTransformer,
 69)
 70from airbyte_cdk.sources.declarative.parsers.manifest_normalizer import (
 71    ManifestNormalizer,
 72)
 73from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import (
 74    ManifestReferenceResolver,
 75)
 76from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
 77    ModelToComponentFactory,
 78)
 79from airbyte_cdk.sources.declarative.resolvers import COMPONENTS_RESOLVER_TYPE_MAPPING
 80from airbyte_cdk.sources.declarative.spec.spec import Spec
 81from airbyte_cdk.sources.declarative.types import Config, ConnectionDefinition
 82from airbyte_cdk.sources.message.concurrent_repository import ConcurrentMessageRepository
 83from airbyte_cdk.sources.message.repository import InMemoryMessageRepository
 84from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 85from airbyte_cdk.sources.streams.concurrent.partitions.types import QueueItem
 86from airbyte_cdk.sources.utils.slice_logger import (
 87    AlwaysLogSliceLogger,
 88    DebugSliceLogger,
 89    SliceLogger,
 90)
 91from airbyte_cdk.utils.stream_status_utils import as_airbyte_message
 92from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 93
 94
 95@dataclass
 96class TestLimits:
 97    __test__: ClassVar[bool] = False  # Tell Pytest this is not a Pytest class, despite its name
 98
 99    DEFAULT_MAX_PAGES_PER_SLICE: ClassVar[int] = 5
100    DEFAULT_MAX_SLICES: ClassVar[int] = 5
101    DEFAULT_MAX_RECORDS: ClassVar[int] = 100
102    DEFAULT_MAX_STREAMS: ClassVar[int] = 100
103
104    max_records: int = field(default=DEFAULT_MAX_RECORDS)
105    max_pages_per_slice: int = field(default=DEFAULT_MAX_PAGES_PER_SLICE)
106    max_slices: int = field(default=DEFAULT_MAX_SLICES)
107    max_streams: int = field(default=DEFAULT_MAX_STREAMS)
108
109
110def _get_declarative_component_schema() -> Dict[str, Any]:
111    try:
112        raw_component_schema = pkgutil.get_data(
113            "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
114        )
115        if raw_component_schema is not None:
116            declarative_component_schema = yaml.load(raw_component_schema, Loader=yaml.SafeLoader)
117            return declarative_component_schema  # type: ignore
118        else:
119            raise RuntimeError(
120                "Failed to read manifest component json schema required for deduplication"
121            )
122    except FileNotFoundError as e:
123        raise FileNotFoundError(
124            f"Failed to read manifest component json schema required for deduplication: {e}"
125        )
126
127
128class ConcurrentDeclarativeSource(Source):
129    # By default, we defer to a value of 2. A value lower than could cause a PartitionEnqueuer to be stuck in a state of deadlock
130    # because it has hit the limit of futures but not partition reader is consuming them.
131    _LOWEST_SAFE_CONCURRENCY_LEVEL = 2
132
133    def __init__(
134        self,
135        catalog: Optional[ConfiguredAirbyteCatalog] = None,
136        config: Optional[Mapping[str, Any]] = None,
137        state: Optional[List[AirbyteStateMessage]] = None,
138        *,
139        source_config: ConnectionDefinition,
140        debug: bool = False,
141        emit_connector_builder_messages: bool = False,
142        migrate_manifest: bool = False,
143        normalize_manifest: bool = False,
144        limits: Optional[TestLimits] = None,
145        config_path: Optional[str] = None,
146        **kwargs: Any,
147    ) -> None:
148        self.logger = logging.getLogger(f"airbyte.{self.name}")
149
150        self._limits = limits
151
152        # todo: We could remove state from initialization. Now that streams are grouped during the read(), a source
153        #  no longer needs to store the original incoming state. But maybe there's an edge case?
154        self._connector_state_manager = ConnectorStateManager(state=state)  # type: ignore  # state is always in the form of List[AirbyteStateMessage]. The ConnectorStateManager should use generics, but this can be done later
155
156        # We set a maxsize to for the main thread to process record items when the queue size grows. This assumes that there are less
157        # threads generating partitions that than are max number of workers. If it weren't the case, we could have threads only generating
158        # partitions which would fill the queue. This number is arbitrarily set to 10_000 but will probably need to be changed given more
159        # information and might even need to be configurable depending on the source
160        queue: Queue[QueueItem] = Queue(maxsize=10_000)
161        message_repository = InMemoryMessageRepository(
162            Level.DEBUG if emit_connector_builder_messages else Level.INFO
163        )
164
165        # To reduce the complexity of the concurrent framework, we are not enabling RFR with synthetic
166        # cursors. We do this by no longer automatically instantiating RFR cursors when converting
167        # the declarative models into runtime components. Concurrent sources will continue to checkpoint
168        # incremental streams running in full refresh.
169        component_factory = ModelToComponentFactory(
170            emit_connector_builder_messages=emit_connector_builder_messages,
171            message_repository=ConcurrentMessageRepository(queue, message_repository),
172            configured_catalog=catalog,
173            connector_state_manager=self._connector_state_manager,
174            max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
175            limit_pages_fetched_per_slice=limits.max_pages_per_slice if limits else None,
176            limit_slices_fetched=limits.max_slices if limits else None,
177            disable_retries=True if limits else False,
178            disable_cache=True if limits else False,
179        )
180
181        self._should_normalize = normalize_manifest
182        self._should_migrate = migrate_manifest
183        self._declarative_component_schema = _get_declarative_component_schema()
184        # If custom components are needed, locate and/or register them.
185        self.components_module: ModuleType | None = get_registered_components_module(config=config)
186        # set additional attributes
187        self._debug = debug
188        self._emit_connector_builder_messages = emit_connector_builder_messages
189        self._constructor = (
190            component_factory
191            if component_factory
192            else ModelToComponentFactory(
193                emit_connector_builder_messages=emit_connector_builder_messages,
194                max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
195            )
196        )
197
198        self._message_repository = self._constructor.get_message_repository()
199        self._slice_logger: SliceLogger = (
200            AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
201        )
202
203        # resolve all components in the manifest
204        self._source_config = self._pre_process_manifest(dict(source_config))
205        # validate resolved manifest against the declarative component schema
206        self._validate_source()
207        # apply additional post-processing to the manifest
208        self._post_process_manifest()
209
210        spec: Optional[Mapping[str, Any]] = self._source_config.get("spec")
211        self._spec_component: Optional[Spec] = (
212            self._constructor.create_component(SpecModel, spec, dict()) if spec else None
213        )
214        self._config = self._migrate_and_transform_config(config_path, config) or {}
215
216        concurrency_level_from_manifest = self._source_config.get("concurrency_level")
217        if concurrency_level_from_manifest:
218            concurrency_level_component = self._constructor.create_component(
219                model_type=ConcurrencyLevelModel,
220                component_definition=concurrency_level_from_manifest,
221                config=config or {},
222            )
223            if not isinstance(concurrency_level_component, ConcurrencyLevel):
224                raise ValueError(
225                    f"Expected to generate a ConcurrencyLevel component, but received {concurrency_level_component.__class__}"
226                )
227
228            concurrency_level = concurrency_level_component.get_concurrency_level()
229            initial_number_of_partitions_to_generate = max(
230                concurrency_level // 2, 1
231            )  # Partition_generation iterates using range based on this value. If this is floored to zero we end up in a dead lock during start up
232        else:
233            concurrency_level = self._LOWEST_SAFE_CONCURRENCY_LEVEL
234            initial_number_of_partitions_to_generate = self._LOWEST_SAFE_CONCURRENCY_LEVEL // 2
235
236        self._concurrent_source = ConcurrentSource.create(
237            num_workers=concurrency_level,
238            initial_number_of_partitions_to_generate=initial_number_of_partitions_to_generate,
239            logger=self.logger,
240            slice_logger=self._slice_logger,
241            queue=queue,
242            message_repository=self._message_repository,
243        )
244
245    def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
246        """
247        Preprocesses the provided manifest dictionary by resolving any manifest references.
248
249        This method modifies the input manifest in place, resolving references using the
250        ManifestReferenceResolver to ensure all references within the manifest are properly handled.
251
252        Args:
253            manifest (Dict[str, Any]): The manifest dictionary to preprocess and resolve references in.
254
255        Returns:
256            None
257        """
258        # For ease of use we don't require the type to be specified at the top level manifest, but it should be included during processing
259        manifest = self._fix_source_type(manifest)
260        # Resolve references in the manifest
261        resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
262        # Propagate types and parameters throughout the manifest
263        propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters(
264            "", resolved_manifest, {}
265        )
266
267        return propagated_manifest
268
269    def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
270        """
271        Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest.
272        """
273        if "type" not in manifest:
274            manifest["type"] = "DeclarativeSource"
275
276        return manifest
277
278    def _post_process_manifest(self) -> None:
279        """
280        Post-processes the manifest after validation.
281        This method is responsible for any additional modifications or transformations needed
282        after the manifest has been validated and before it is used in the source.
283        """
284        # apply manifest migration, if required
285        self._migrate_manifest()
286        # apply manifest normalization, if required
287        self._normalize_manifest()
288
289    def _migrate_manifest(self) -> None:
290        """
291        This method is used to migrate the manifest. It should be called after the manifest has been validated.
292        The migration is done in place, so the original manifest is modified.
293
294        The original manifest is returned if any error occurs during migration.
295        """
296        if self._should_migrate:
297            manifest_migrator = ManifestMigrationHandler(self._source_config)
298            self._source_config = manifest_migrator.apply_migrations()
299            # validate migrated manifest against the declarative component schema
300            self._validate_source()
301
302    def _normalize_manifest(self) -> None:
303        """
304        This method is used to normalize the manifest. It should be called after the manifest has been validated.
305
306        Connector Builder UI rendering requires the manifest to be in a specific format.
307         - references have been resolved
308         - the commonly used definitions are extracted to the `definitions.linked.*`
309        """
310        if self._should_normalize:
311            normalizer = ManifestNormalizer(self._source_config, self._declarative_component_schema)
312            self._source_config = normalizer.normalize()
313
314    def _validate_source(self) -> None:
315        """
316        Validates the connector manifest against the declarative component schema
317        """
318
319        try:
320            validate(self._source_config, self._declarative_component_schema)
321        except ValidationError as e:
322            raise ValidationError(
323                "Validation against json schema defined in declarative_component_schema.yaml schema failed"
324            ) from e
325
326    def _migrate_and_transform_config(
327        self,
328        config_path: Optional[str],
329        config: Optional[Config],
330    ) -> Optional[Config]:
331        if not config:
332            return None
333        if not self._spec_component:
334            return config
335        mutable_config = dict(config)
336        self._spec_component.migrate_config(mutable_config)
337        if mutable_config != config:
338            if config_path:
339                with open(config_path, "w") as f:
340                    json.dump(mutable_config, f)
341            control_message = create_connector_config_control_message(mutable_config)
342            print(orjson.dumps(AirbyteMessageSerializer.dump(control_message)).decode())
343        self._spec_component.transform_config(mutable_config)
344        return mutable_config
345
346    def configure(self, config: Mapping[str, Any], temp_dir: str) -> Mapping[str, Any]:
347        config = self._config or config
348        return super().configure(config, temp_dir)
349
350    @property
351    def resolved_manifest(self) -> Mapping[str, Any]:
352        """
353        Returns the resolved manifest configuration for the source.
354
355        This property provides access to the internal source configuration as a mapping,
356        which contains all settings and parameters required to define the source's behavior.
357
358        Returns:
359            Mapping[str, Any]: The resolved source configuration manifest.
360        """
361        return self._source_config
362
363    def deprecation_warnings(self) -> List[ConnectorBuilderLogMessage]:
364        return self._constructor.get_model_deprecations()
365
366    def read(
367        self,
368        logger: logging.Logger,
369        config: Mapping[str, Any],
370        catalog: ConfiguredAirbyteCatalog,
371        state: Optional[List[AirbyteStateMessage]] = None,
372    ) -> Iterator[AirbyteMessage]:
373        selected_concurrent_streams = self._select_streams(
374            streams=self.streams(config=self._config),  # type: ignore  # We are migrating away from the DeclarativeStream implementation and streams() only returns the concurrent-compatible AbstractStream. To preserve compatibility, we retain the existing method interface
375            configured_catalog=catalog,
376        )
377
378        # It would appear that passing in an empty set of streams causes an infinite loop in ConcurrentReadProcessor.
379        # This is also evident in concurrent_source_adapter.py so I'll leave this out of scope to fix for now
380        if len(selected_concurrent_streams) > 0:
381            yield from self._concurrent_source.read(selected_concurrent_streams)
382
383    def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog:
384        return AirbyteCatalog(
385            streams=[stream.as_airbyte_stream() for stream in self.streams(config=self._config)]
386        )
387
388    # todo: add PR comment about whether we can change the signature to List[AbstractStream]
389    def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]:  # type: ignore  # we are migrating away from the AbstractSource and are expecting that this will only be called by ConcurrentDeclarativeSource or the Connector Builder
390        """
391        The `streams` method is used as part of the AbstractSource in the following cases:
392        * ConcurrentDeclarativeSource.check -> ManifestDeclarativeSource.check -> AbstractSource.check -> DeclarativeSource.check_connection -> CheckStream.check_connection -> streams
393        * ConcurrentDeclarativeSource.read -> AbstractSource.read -> streams (note that we filter for a specific catalog which excludes concurrent streams so not all streams actually read from all the streams returned by `streams`)
394        Note that `super.streams(config)` is also called when splitting the streams between concurrent or not in `_group_streams`.
395
396        In both case, we will assume that calling the DeclarativeStream is perfectly fine as the result for these is the same regardless of if it is a DeclarativeStream or a DefaultStream (concurrent). This should simply be removed once we have moved away from the mentioned code paths above.
397        """
398
399        if self._spec_component:
400            self._spec_component.validate_config(self._config)
401
402        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
403
404        api_budget_model = self._source_config.get("api_budget")
405        if api_budget_model:
406            self._constructor.set_api_budget(api_budget_model, self._config)
407
408        source_streams = [
409            self._constructor.create_component(
410                (
411                    StateDelegatingStreamModel
412                    if stream_config.get("type") == StateDelegatingStreamModel.__name__
413                    else DeclarativeStreamModel
414                ),
415                stream_config,
416                self._config,
417                emit_connector_builder_messages=self._emit_connector_builder_messages,
418            )
419            for stream_config in self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
420        ]
421        return source_streams
422
423    @staticmethod
424    def _initialize_cache_for_parent_streams(
425        stream_configs: List[Dict[str, Any]],
426    ) -> List[Dict[str, Any]]:
427        """Enable caching for parent streams unless explicitly disabled.
428
429        Caching is enabled by default for parent streams to optimize performance when the same
430        parent data is needed by multiple child streams. However, explicit `use_cache: false`
431        settings are respected for streams that cannot use caching (e.g., scroll-based pagination
432        APIs where caching causes duplicate records).
433        """
434        parent_streams = set()
435
436        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
437            """Set use_cache to True only if not explicitly disabled."""
438            if requester.get("use_cache") is not False:
439                requester["use_cache"] = True
440
441        def update_with_cache_parent_configs(
442            parent_configs: list[dict[str, Any]],
443        ) -> None:
444            for parent_config in parent_configs:
445                parent_streams.add(parent_config["stream"]["name"])
446                if parent_config["stream"]["type"] == "StateDelegatingStream":
447                    _set_cache_if_not_disabled(
448                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
449                    )
450                    _set_cache_if_not_disabled(
451                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
452                    )
453                else:
454                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
455
456        for stream_config in stream_configs:
457            if stream_config.get("incremental_sync", {}).get("parent_stream"):
458                parent_streams.add(stream_config["incremental_sync"]["parent_stream"]["name"])
459                _set_cache_if_not_disabled(
460                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
461                )
462
463            elif stream_config.get("retriever", {}).get("partition_router", {}):
464                partition_router = stream_config["retriever"]["partition_router"]
465
466                if isinstance(partition_router, dict) and partition_router.get(
467                    "parent_stream_configs"
468                ):
469                    update_with_cache_parent_configs(partition_router["parent_stream_configs"])
470                elif isinstance(partition_router, list):
471                    for router in partition_router:
472                        if router.get("parent_stream_configs"):
473                            update_with_cache_parent_configs(router["parent_stream_configs"])
474
475        for stream_config in stream_configs:
476            if stream_config["name"] in parent_streams:
477                if stream_config["type"] == "StateDelegatingStream":
478                    _set_cache_if_not_disabled(
479                        stream_config["full_refresh_stream"]["retriever"]["requester"]
480                    )
481                    _set_cache_if_not_disabled(
482                        stream_config["incremental_stream"]["retriever"]["requester"]
483                    )
484                else:
485                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
486        return stream_configs
487
488    def spec(self, logger: logging.Logger) -> ConnectorSpecification:
489        """
490        Returns the connector specification (spec) as defined in the Airbyte Protocol. The spec is an object describing the possible
491        configurations (e.g: username and password) which can be configured when running this connector. For low-code connectors, this
492        will first attempt to load the spec from the manifest's spec block, otherwise it will load it from "spec.yaml" or "spec.json"
493        in the project root.
494        """
495        return (
496            self._spec_component.generate_spec() if self._spec_component else super().spec(logger)
497        )
498
499    def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
500        check = self._source_config.get("check")
501        if not check:
502            raise ValueError(f"Missing 'check' component definition within the manifest.")
503
504        if "type" not in check:
505            check["type"] = "CheckStream"
506        connection_checker = self._constructor.create_component(
507            COMPONENTS_CHECKER_TYPE_MAPPING[check["type"]],
508            check,
509            dict(),
510            emit_connector_builder_messages=self._emit_connector_builder_messages,
511        )
512        if not isinstance(connection_checker, ConnectionChecker):
513            raise ValueError(
514                f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}"
515            )
516
517        check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
518        if not check_succeeded:
519            return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error))
520        return AirbyteConnectionStatus(status=Status.SUCCEEDED)
521
522    @property
523    def dynamic_streams(self) -> List[Dict[str, Any]]:
524        return self._dynamic_stream_configs(
525            manifest=self._source_config,
526            with_dynamic_stream_name=True,
527        )
528
529    def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
530        # This has a warning flag for static, but after we finish part 4 we'll replace manifest with self._source_config
531        stream_configs = []
532        for current_stream_config in manifest.get("streams", []):
533            if (
534                "type" in current_stream_config
535                and current_stream_config["type"] == "ConditionalStreams"
536            ):
537                interpolated_boolean = InterpolatedBoolean(
538                    condition=current_stream_config.get("condition"),
539                    parameters={},
540                )
541
542                if interpolated_boolean.eval(config=self._config):
543                    stream_configs.extend(current_stream_config.get("streams", []))
544            else:
545                if "type" not in current_stream_config:
546                    current_stream_config["type"] = "DeclarativeStream"
547                stream_configs.append(current_stream_config)
548        return stream_configs
549
550    def _dynamic_stream_configs(
551        self,
552        manifest: Mapping[str, Any],
553        with_dynamic_stream_name: Optional[bool] = None,
554    ) -> List[Dict[str, Any]]:
555        dynamic_stream_definitions: List[Dict[str, Any]] = manifest.get("dynamic_streams", [])
556        dynamic_stream_configs: List[Dict[str, Any]] = []
557        seen_dynamic_streams: Set[str] = set()
558
559        for dynamic_definition_index, dynamic_definition in enumerate(dynamic_stream_definitions):
560            components_resolver_config = dynamic_definition["components_resolver"]
561
562            if not components_resolver_config:
563                raise ValueError(
564                    f"Missing 'components_resolver' in dynamic definition: {dynamic_definition}"
565                )
566
567            resolver_type = components_resolver_config.get("type")
568            if not resolver_type:
569                raise ValueError(
570                    f"Missing 'type' in components resolver configuration: {components_resolver_config}"
571                )
572
573            if resolver_type not in COMPONENTS_RESOLVER_TYPE_MAPPING:
574                raise ValueError(
575                    f"Invalid components resolver type '{resolver_type}'. "
576                    f"Expected one of {list(COMPONENTS_RESOLVER_TYPE_MAPPING.keys())}."
577                )
578
579            if "retriever" in components_resolver_config:
580                components_resolver_config["retriever"]["requester"]["use_cache"] = True
581
582            # Create a resolver for dynamic components based on type
583            if resolver_type == "HttpComponentsResolver":
584                components_resolver = self._constructor.create_component(
585                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
586                    component_definition=components_resolver_config,
587                    config=self._config,
588                    stream_name=dynamic_definition.get("name"),
589                )
590            else:
591                components_resolver = self._constructor.create_component(
592                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
593                    component_definition=components_resolver_config,
594                    config=self._config,
595                )
596
597            stream_template_config = dynamic_definition["stream_template"]
598
599            for dynamic_stream in components_resolver.resolve_components(
600                stream_template_config=stream_template_config
601            ):
602                # Get the use_parent_parameters configuration from the dynamic definition
603                # Default to True for backward compatibility, since connectors were already using it by default when this param was added
604                use_parent_parameters = dynamic_definition.get("use_parent_parameters", True)
605
606                dynamic_stream = {
607                    **ManifestComponentTransformer().propagate_types_and_parameters(
608                        "", dynamic_stream, {}, use_parent_parameters=use_parent_parameters
609                    )
610                }
611
612                if "type" not in dynamic_stream:
613                    dynamic_stream["type"] = "DeclarativeStream"
614
615                # Ensure that each stream is created with a unique name
616                name = dynamic_stream.get("name")
617
618                if with_dynamic_stream_name:
619                    dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
620                        "name", f"dynamic_stream_{dynamic_definition_index}"
621                    )
622
623                if not isinstance(name, str):
624                    raise ValueError(
625                        f"Expected stream name {name} to be a string, got {type(name)}."
626                    )
627
628                if name in seen_dynamic_streams:
629                    error_message = f"Dynamic streams list contains a duplicate name: {name}. Please contact Airbyte Support."
630                    failure_type = FailureType.system_error
631
632                    if resolver_type == "ConfigComponentsResolver":
633                        error_message = f"Dynamic streams list contains a duplicate name: {name}. Please check your configuration."
634                        failure_type = FailureType.config_error
635
636                    raise AirbyteTracedException(
637                        message=error_message,
638                        internal_message=error_message,
639                        failure_type=failure_type,
640                    )
641
642                seen_dynamic_streams.add(name)
643                dynamic_stream_configs.append(dynamic_stream)
644
645        return dynamic_stream_configs
646
647    def _select_streams(
648        self, streams: List[AbstractStream], configured_catalog: ConfiguredAirbyteCatalog
649    ) -> List[AbstractStream]:
650        stream_name_to_instance: Mapping[str, AbstractStream] = {s.name: s for s in streams}
651        abstract_streams: List[AbstractStream] = []
652        for configured_stream in configured_catalog.streams:
653            stream_instance = stream_name_to_instance.get(configured_stream.stream.name)
654            if stream_instance:
655                abstract_streams.append(stream_instance)
656            else:
657                # Previous behavior in the legacy synchronous CDK was to also raise an error TRACE message if
658                # the source was configured with raise_exception_on_missing_stream=True. This was used on very
659                # few sources like facebook-marketing and google-ads. We decided not to port this feature over,
660                # but we can do so if we feel it necessary. With the current behavior,we should still result
661                # in a partial failure since missing streams will be marked as INCOMPLETE.
662                self._message_repository.emit_message(
663                    as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE)
664                )
665        return abstract_streams