airbyte_cdk.sources.declarative.concurrent_declarative_source

   1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
   2
   3import json
   4import logging
   5import pkgutil
   6from contextlib import contextmanager
   7from copy import deepcopy
   8from dataclasses import dataclass, field
   9from queue import Queue
  10from types import ModuleType
  11from typing import (
  12    Any,
  13    ClassVar,
  14    Dict,
  15    Iterator,
  16    List,
  17    Mapping,
  18    Optional,
  19    Set,
  20)
  21
  22import orjson
  23from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor
  24from jsonschema.exceptions import ValidationError, best_match
  25from jsonschema.validators import validator_for
  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.partition_routers.cartesian_product_stream_slicer import (
  80    CartesianProductStreamSlicer,
  81)
  82from airbyte_cdk.sources.declarative.partition_routers.grouping_partition_router import (
  83    GroupingPartitionRouter,
  84)
  85from airbyte_cdk.sources.declarative.partition_routers.substream_partition_router import (
  86    SubstreamPartitionRouter,
  87)
  88from airbyte_cdk.sources.declarative.partition_routers.union_partition_router import (
  89    UnionPartitionRouter,
  90)
  91from airbyte_cdk.sources.declarative.resolvers import COMPONENTS_RESOLVER_TYPE_MAPPING
  92from airbyte_cdk.sources.declarative.spec.spec import Spec
  93from airbyte_cdk.sources.declarative.types import Config, ConnectionDefinition
  94from airbyte_cdk.sources.message.concurrent_repository import ConcurrentMessageRepository
  95from airbyte_cdk.sources.message.repository import InMemoryMessageRepository
  96from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
  97from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream
  98from airbyte_cdk.sources.streams.concurrent.partitions.types import QueueItem
  99from airbyte_cdk.sources.utils.slice_logger import (
 100    AlwaysLogSliceLogger,
 101    DebugSliceLogger,
 102    SliceLogger,
 103)
 104from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, get_secrets
 105from airbyte_cdk.utils.stream_status_utils import as_airbyte_message
 106from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 107from airbyte_cdk.utils.yaml_loader import safe_load_yaml
 108
 109
 110@dataclass
 111class TestLimits:
 112    __test__: ClassVar[bool] = False  # Tell Pytest this is not a Pytest class, despite its name
 113
 114    DEFAULT_MAX_PAGES_PER_SLICE: ClassVar[int] = 5
 115    DEFAULT_MAX_SLICES: ClassVar[int] = 5
 116    DEFAULT_MAX_RECORDS: ClassVar[int] = 100
 117    DEFAULT_MAX_STREAMS: ClassVar[int] = 100
 118
 119    max_records: int = field(default=DEFAULT_MAX_RECORDS)
 120    max_pages_per_slice: int = field(default=DEFAULT_MAX_PAGES_PER_SLICE)
 121    max_slices: int = field(default=DEFAULT_MAX_SLICES)
 122    max_streams: int = field(default=DEFAULT_MAX_STREAMS)
 123
 124
 125def get_declarative_component_schema() -> Dict[str, Any]:
 126    try:
 127        raw_component_schema = pkgutil.get_data(
 128            "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
 129        )
 130        if raw_component_schema is not None:
 131            declarative_component_schema = safe_load_yaml(raw_component_schema)
 132            return declarative_component_schema  # type: ignore
 133        else:
 134            raise RuntimeError(
 135                "Failed to read manifest component json schema required for deduplication"
 136            )
 137    except FileNotFoundError as e:
 138        raise FileNotFoundError(
 139            f"Failed to read manifest component json schema required for deduplication: {e}"
 140        )
 141
 142
 143def validate_manifest(manifest: Mapping[str, Any], schema: Mapping[str, Any]) -> None:
 144    """Validate a manifest against the shipped component schema.
 145
 146    Unit tests check the schema's metaschema, so valid manifests skip `check_schema`. On failure,
 147    check the schema first to surface `SchemaError`, while preserving `jsonschema.validate`'s
 148    best-match error as the cause with a traceback.
 149    """
 150    validator_class = validator_for(schema)
 151    try:
 152        error = best_match(validator_class(schema).iter_errors(manifest))
 153        if error is not None:
 154            raise error
 155    except ValidationError as e:
 156        validator_class.check_schema(schema)
 157        raise ValidationError(
 158            "Validation against json schema defined in declarative_component_schema.yaml schema failed"
 159        ) from e
 160    except Exception:
 161        validator_class.check_schema(schema)
 162        raise
 163
 164
 165class ConcurrentDeclarativeSource(Source):
 166    # 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
 167    # because it has hit the limit of futures but not partition reader is consuming them.
 168    _LOWEST_SAFE_CONCURRENCY_LEVEL = 2
 169
 170    # Component types that hold the connector config and can write it back. `refresh_token_updater` is
 171    # declared on `OAuthAuthenticator` alone; `CustomAuthenticator` is the type the transformer injects
 172    # for a `class_name` component.
 173    _CONFIG_PERSISTING_AUTHENTICATOR_TYPES = frozenset(
 174        {"OAuthAuthenticator", "CustomAuthenticator"}
 175    )
 176
 177    def __init__(
 178        self,
 179        catalog: Optional[ConfiguredAirbyteCatalog] = None,
 180        config: Optional[Mapping[str, Any]] = None,
 181        state: Optional[List[AirbyteStateMessage]] = None,
 182        *,
 183        source_config: ConnectionDefinition,
 184        debug: bool = False,
 185        emit_connector_builder_messages: bool = False,
 186        migrate_manifest: bool = False,
 187        normalize_manifest: bool = False,
 188        limits: Optional[TestLimits] = None,
 189        config_path: Optional[str] = None,
 190        custom_components_trusted: bool = True,
 191        **kwargs: Any,
 192    ) -> None:
 193        self.logger = logging.getLogger(f"airbyte.{self.name}")
 194
 195        self._limits = limits
 196
 197        # todo: We could remove state from initialization. Now that streams are grouped during the read(), a source
 198        #  no longer needs to store the original incoming state. But maybe there's an edge case?
 199        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
 200
 201        # We set a maxsize to for the main thread to process record items when the queue size grows. This assumes that there are less
 202        # threads generating partitions that than are max number of workers. If it weren't the case, we could have threads only generating
 203        # partitions which would fill the queue. This number is arbitrarily set to 10_000 but will probably need to be changed given more
 204        # information and might even need to be configurable depending on the source
 205        queue: Queue[QueueItem] = Queue(maxsize=10_000)
 206        message_repository = InMemoryMessageRepository(
 207            Level.DEBUG if emit_connector_builder_messages else Level.INFO
 208        )
 209
 210        # To reduce the complexity of the concurrent framework, we are not enabling RFR with synthetic
 211        # cursors. We do this by no longer automatically instantiating RFR cursors when converting
 212        # the declarative models into runtime components. Concurrent sources will continue to checkpoint
 213        # incremental streams running in full refresh.
 214        component_factory = ModelToComponentFactory(
 215            custom_components_trusted=custom_components_trusted,
 216            emit_connector_builder_messages=emit_connector_builder_messages,
 217            message_repository=ConcurrentMessageRepository(queue, message_repository),
 218            configured_catalog=catalog,
 219            connector_state_manager=self._connector_state_manager,
 220            max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
 221            limit_pages_fetched_per_slice=limits.max_pages_per_slice if limits else None,
 222            limit_slices_fetched=limits.max_slices if limits else None,
 223            disable_retries=True if limits else False,
 224            disable_cache=True if limits else False,
 225        )
 226
 227        self._should_normalize = normalize_manifest
 228        self._should_migrate = migrate_manifest
 229        self._declarative_component_schema = get_declarative_component_schema()
 230        # If custom components are needed, locate and/or register them.
 231        self.components_module: ModuleType | None = get_registered_components_module(config=config)
 232        # set additional attributes
 233        self._debug = debug
 234        self._emit_connector_builder_messages = emit_connector_builder_messages
 235        self._constructor = (
 236            component_factory
 237            if component_factory
 238            else ModelToComponentFactory(
 239                custom_components_trusted=custom_components_trusted,
 240                emit_connector_builder_messages=emit_connector_builder_messages,
 241                max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
 242            )
 243        )
 244
 245        self._message_repository = self._constructor.get_message_repository()
 246        self._slice_logger: SliceLogger = (
 247            AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
 248        )
 249
 250        # resolve all components in the manifest
 251        self._source_config = self._pre_process_manifest(dict(source_config))
 252        # validate resolved manifest against the declarative component schema
 253        self._validate_source()
 254        # apply additional post-processing to the manifest
 255        self._post_process_manifest()
 256
 257        spec: Optional[Mapping[str, Any]] = self._source_config.get("spec")
 258        self._spec_component: Optional[Spec] = (
 259            self._constructor.create_component(SpecModel, spec, dict()) if spec else None
 260        )
 261        self._config = self._migrate_and_transform_config(config_path, config) or {}
 262        # `check` may temporarily overlay values onto `self._config` (see
 263        # `_config_overridden_for_check`). The manifest's `config_validations` express intent about the
 264        # config as supplied, so they must run against it rather than against a check-time overlay.
 265        self._config_for_validation = self._config
 266
 267        concurrency_level_from_manifest = self._source_config.get("concurrency_level")
 268        if concurrency_level_from_manifest:
 269            concurrency_level_component = self._constructor.create_component(
 270                model_type=ConcurrencyLevelModel,
 271                component_definition=concurrency_level_from_manifest,
 272                config=self._config,
 273            )
 274            if not isinstance(concurrency_level_component, ConcurrencyLevel):
 275                raise ValueError(
 276                    f"Expected to generate a ConcurrencyLevel component, but received {concurrency_level_component.__class__}"
 277                )
 278
 279            concurrency_level = concurrency_level_component.get_concurrency_level()
 280            initial_number_of_partitions_to_generate = max(
 281                concurrency_level // 2, 1
 282            )  # 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
 283        else:
 284            concurrency_level = self._LOWEST_SAFE_CONCURRENCY_LEVEL
 285            initial_number_of_partitions_to_generate = self._LOWEST_SAFE_CONCURRENCY_LEVEL // 2
 286
 287        self._concurrent_source = ConcurrentSource.create(
 288            num_workers=concurrency_level,
 289            initial_number_of_partitions_to_generate=initial_number_of_partitions_to_generate,
 290            logger=self.logger,
 291            slice_logger=self._slice_logger,
 292            queue=queue,
 293            message_repository=self._message_repository,
 294        )
 295
 296    def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
 297        """
 298        Preprocesses the provided manifest dictionary by resolving any manifest references.
 299
 300        This method modifies the input manifest in place, resolving references using the
 301        ManifestReferenceResolver to ensure all references within the manifest are properly handled.
 302
 303        Args:
 304            manifest (Dict[str, Any]): The manifest dictionary to preprocess and resolve references in.
 305
 306        Returns:
 307            None
 308        """
 309        # 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
 310        manifest = self._fix_source_type(manifest)
 311        # Resolve references in the manifest
 312        resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
 313        # Propagate types and parameters throughout the manifest
 314        propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters(
 315            "", resolved_manifest, {}
 316        )
 317
 318        return propagated_manifest
 319
 320    def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
 321        """
 322        Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest.
 323        """
 324        if "type" not in manifest:
 325            manifest["type"] = "DeclarativeSource"
 326
 327        return manifest
 328
 329    def _post_process_manifest(self) -> None:
 330        """
 331        Post-processes the manifest after validation.
 332        This method is responsible for any additional modifications or transformations needed
 333        after the manifest has been validated and before it is used in the source.
 334        """
 335        # apply manifest migration, if required
 336        self._migrate_manifest()
 337        # apply manifest normalization, if required
 338        self._normalize_manifest()
 339
 340    def _migrate_manifest(self) -> None:
 341        """
 342        This method is used to migrate the manifest. It should be called after the manifest has been validated.
 343        The migration is done in place, so the original manifest is modified.
 344
 345        The original manifest is returned if any error occurs during migration.
 346        """
 347        if self._should_migrate:
 348            manifest_migrator = ManifestMigrationHandler(self._source_config)
 349            self._source_config = manifest_migrator.apply_migrations()
 350            # validate migrated manifest against the declarative component schema
 351            self._validate_source()
 352
 353    def _normalize_manifest(self) -> None:
 354        """
 355        This method is used to normalize the manifest. It should be called after the manifest has been validated.
 356
 357        Connector Builder UI rendering requires the manifest to be in a specific format.
 358         - references have been resolved
 359         - the commonly used definitions are extracted to the `definitions.linked.*`
 360        """
 361        if self._should_normalize:
 362            normalizer = ManifestNormalizer(self._source_config, self._declarative_component_schema)
 363            self._source_config = normalizer.normalize()
 364
 365    def _validate_source(self) -> None:
 366        """
 367        Validates the connector manifest against the declarative component schema
 368        """
 369
 370        validate_manifest(self._source_config, self._declarative_component_schema)
 371
 372    def _migrate_and_transform_config(
 373        self,
 374        config_path: Optional[str],
 375        config: Optional[Config],
 376    ) -> Optional[Config]:
 377        if not config:
 378            return None
 379        if not self._spec_component:
 380            return config
 381        mutable_config = dict(config)
 382        self._spec_component.migrate_config(mutable_config)
 383        if mutable_config != config:
 384            if config_path:
 385                with open(config_path, "w") as f:
 386                    json.dump(mutable_config, f)
 387            control_message = create_connector_config_control_message(mutable_config)
 388            print(orjson.dumps(AirbyteMessageSerializer.dump(control_message)).decode())
 389        self._spec_component.transform_config(mutable_config)
 390        return mutable_config
 391
 392    def configure(self, config: Mapping[str, Any], temp_dir: str) -> Mapping[str, Any]:
 393        config = self._config or config
 394        return super().configure(config, temp_dir)
 395
 396    @property
 397    def resolved_manifest(self) -> Mapping[str, Any]:
 398        """
 399        Returns the resolved manifest configuration for the source.
 400
 401        This property provides access to the internal source configuration as a mapping,
 402        which contains all settings and parameters required to define the source's behavior.
 403
 404        Returns:
 405            Mapping[str, Any]: The resolved source configuration manifest.
 406        """
 407        return self._source_config
 408
 409    def deprecation_warnings(self) -> List[ConnectorBuilderLogMessage]:
 410        return self._constructor.get_model_deprecations()
 411
 412    def read(
 413        self,
 414        logger: logging.Logger,
 415        config: Mapping[str, Any],
 416        catalog: ConfiguredAirbyteCatalog,
 417        state: Optional[List[AirbyteStateMessage]] = None,
 418    ) -> Iterator[AirbyteMessage]:
 419        selected_concurrent_streams = self._select_streams(
 420            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
 421            configured_catalog=catalog,
 422        )
 423
 424        # It would appear that passing in an empty set of streams causes an infinite loop in ConcurrentReadProcessor.
 425        # This is also evident in concurrent_source_adapter.py so I'll leave this out of scope to fix for now
 426        if len(selected_concurrent_streams) > 0:
 427            yield from self._concurrent_source.read(selected_concurrent_streams)
 428
 429    def discover(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCatalog:
 430        return AirbyteCatalog(
 431            streams=[stream.as_airbyte_stream() for stream in self.streams(config=self._config)]
 432        )
 433
 434    # todo: add PR comment about whether we can change the signature to List[AbstractStream]
 435    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
 436        """
 437        The `streams` method is used as part of the AbstractSource in the following cases:
 438        * ConcurrentDeclarativeSource.check -> ManifestDeclarativeSource.check -> AbstractSource.check -> DeclarativeSource.check_connection -> CheckStream.check_connection -> streams
 439        * 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`)
 440        Note that `super.streams(config)` is also called when splitting the streams between concurrent or not in `_group_streams`.
 441
 442        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.
 443        """
 444
 445        self._prepare_for_stream_build()
 446        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
 447        return self._create_streams(stream_configs, stream_configs)
 448
 449    def _prepare_for_stream_build(self) -> None:
 450        if self._spec_component:
 451            self._spec_component.validate_config(self._config_for_validation)
 452
 453        api_budget_model = self._source_config.get("api_budget")
 454        if api_budget_model:
 455            self._constructor.set_api_budget(api_budget_model, self._config)
 456
 457    def _create_streams(
 458        self,
 459        selected_configs: List[Dict[str, Any]],
 460        all_configs: List[Dict[str, Any]],
 461    ) -> List[AbstractStream]:
 462        prepared_configs = self._initialize_cache_for_parent_streams(
 463            deepcopy(selected_configs), all_stream_configs=all_configs
 464        )
 465
 466        source_streams = [
 467            self._constructor.create_component(
 468                (
 469                    StateDelegatingStreamModel
 470                    if stream_config.get("type") == StateDelegatingStreamModel.__name__
 471                    else DeclarativeStreamModel
 472                ),
 473                stream_config,
 474                self._config,
 475                emit_connector_builder_messages=self._emit_connector_builder_messages,
 476            )
 477            for stream_config in prepared_configs
 478        ]
 479
 480        top_level_stream_names: Set[str] = set()
 481        for stream_config in all_configs:
 482            stream_name = stream_config.get("name")
 483            if isinstance(stream_name, str):
 484                top_level_stream_names.add(stream_name)
 485        self._apply_stream_groups(source_streams, top_level_stream_names)
 486
 487        return source_streams
 488
 489    def _uses_default_streams(self) -> bool:
 490        """A subclass that overrides `streams()` must keep going through it during check."""
 491        return type(self).streams is ConcurrentDeclarativeSource.streams
 492
 493    def _streams_by_name(self, stream_names: List[str]) -> Optional[List[AbstractStream]]:
 494        """Build the named static streams without resolving or building dynamic streams.
 495
 496        Return None when the names are empty, any name isn't static, or `streams()` is overridden.
 497        """
 498        if not self._uses_default_streams():
 499            return None
 500        try:
 501            static_configs = self._stream_configs(self._source_config)
 502        except Exception:
 503            return None
 504        static_stream_names = {stream_config.get("name") for stream_config in static_configs}
 505        if not stream_names or any(name not in static_stream_names for name in stream_names):
 506            return None
 507
 508        self._prepare_for_stream_build()
 509        selected_names = set(stream_names)
 510        selected_configs = [
 511            stream_config
 512            for stream_config in static_configs
 513            if stream_config.get("name") in selected_names
 514        ]
 515        return self._create_streams(selected_configs, static_configs)
 516
 517    def _first_streams(self, count: int) -> Optional[List[AbstractStream]]:
 518        """Build the first `count` streams, static then dynamic, in `streams()` order.
 519
 520        `count` must be at least 1. Return None when `streams()` is overridden.
 521        """
 522        if not self._uses_default_streams():
 523            return None
 524        if count < 1:
 525            raise ValueError("count must be >= 1")
 526        self._prepare_for_stream_build()
 527        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
 528        return self._create_streams(stream_configs[:count], stream_configs)
 529
 530    def _apply_stream_groups(
 531        self,
 532        streams: List[AbstractStream],
 533        top_level_stream_names: Optional[Set[str]] = None,
 534    ) -> None:
 535        """Set block_simultaneous_read on streams based on the manifest's stream_groups config.
 536
 537        Iterates over the resolved manifest's stream_groups and matches group membership
 538        against actual created stream instances by name. Validates that no stream shares a
 539        group with any of its parent streams, which would cause a deadlock.
 540        """
 541        stream_groups = self._source_config.get("stream_groups", {})
 542        if not stream_groups:
 543            return
 544
 545        # Build stream_name -> group_name mapping from the resolved manifest
 546        stream_name_to_group: Dict[str, str] = {}
 547        for group_name, group_config in stream_groups.items():
 548            for stream_ref in group_config.get("streams", []):
 549                if isinstance(stream_ref, dict):
 550                    stream_name = stream_ref.get("name", "")
 551                    if stream_name:
 552                        stream_name_to_group[stream_name] = group_name
 553
 554        # Validate no stream shares a group with any of its ancestor streams
 555        stream_name_to_instance: Dict[str, AbstractStream] = {s.name: s for s in streams}
 556
 557        def _collect_all_ancestor_names(
 558            stream_name: str, inline_instance: Optional[AbstractStream] = None
 559        ) -> Set[str]:
 560            """Recursively collect all ancestor stream names."""
 561            ancestors: Set[str] = set()
 562            inst = stream_name_to_instance.get(stream_name)
 563            if inst is None and stream_name in (top_level_stream_names or set()):
 564                # A top-level stream that this call did not build: walk the copy inlined in the child
 565                inst = inline_instance
 566            if not isinstance(inst, DefaultStream):
 567                return ancestors
 568            partition_router = inst.get_partition_router()
 569            routers = [partition_router] if partition_router is not None else []
 570            while routers:
 571                router = routers.pop()
 572                if isinstance(router, GroupingPartitionRouter):
 573                    routers.append(router.underlying_partition_router)
 574                elif isinstance(router, UnionPartitionRouter):
 575                    routers.extend(router.partition_routers)
 576                elif isinstance(router, CartesianProductStreamSlicer):
 577                    routers.extend(router.stream_slicers)
 578                elif isinstance(router, SubstreamPartitionRouter):
 579                    for parent_config in router.parent_stream_configs:
 580                        parent_name = parent_config.stream.name
 581                        ancestors.add(parent_name)
 582                        ancestors.update(
 583                            _collect_all_ancestor_names(parent_name, parent_config.stream)
 584                        )
 585            return ancestors
 586
 587        for stream in streams:
 588            if not isinstance(stream, DefaultStream) or stream.name not in stream_name_to_group:
 589                continue
 590            group_name = stream_name_to_group[stream.name]
 591            for ancestor_name in _collect_all_ancestor_names(stream.name):
 592                if stream_name_to_group.get(ancestor_name) == group_name:
 593                    raise ValueError(
 594                        f"Stream '{stream.name}' and its parent stream '{ancestor_name}' "
 595                        f"are both in group '{group_name}'. "
 596                        f"A child stream must not share a group with its parent to avoid deadlock."
 597                    )
 598
 599        # Apply group to matching stream instances
 600        for stream in streams:
 601            if isinstance(stream, DefaultStream) and stream.name in stream_name_to_group:
 602                stream.block_simultaneous_read = stream_name_to_group[stream.name]
 603
 604    @staticmethod
 605    def _parent_stream_names(stream_configs: List[Dict[str, Any]]) -> Set[str]:
 606        parent_stream_names: Set[str] = set()
 607        for stream_config in stream_configs:
 608            parent_stream = stream_config.get("incremental_sync", {}).get("parent_stream")
 609            if parent_stream:
 610                parent_stream_names.add(parent_stream["name"])
 611            elif stream_config.get("retriever", {}).get("partition_router", {}):
 612                partition_router = stream_config["retriever"]["partition_router"]
 613                routers = (
 614                    list(partition_router)
 615                    if isinstance(partition_router, list)
 616                    else [partition_router]
 617                )
 618                while routers:
 619                    router = routers.pop()
 620                    if not isinstance(router, dict):
 621                        continue
 622                    parent_stream_names.update(
 623                        parent_config["stream"]["name"]
 624                        for parent_config in router.get("parent_stream_configs") or []
 625                    )
 626                    routers.extend(router.get("partition_routers") or [])
 627                    if router.get("underlying_partition_router"):
 628                        routers.append(router["underlying_partition_router"])
 629        return parent_stream_names
 630
 631    @staticmethod
 632    def _initialize_cache_for_parent_streams(
 633        stream_configs: List[Dict[str, Any]],
 634        all_stream_configs: Optional[List[Dict[str, Any]]] = None,
 635    ) -> List[Dict[str, Any]]:
 636        """Enable caching for parent streams unless explicitly disabled.
 637
 638        Caching is enabled by default for parent streams to optimize performance when the same
 639        parent data is needed by multiple child streams. However, explicit `use_cache: false`
 640        settings are respected for streams that cannot use caching (e.g., scroll-based pagination
 641        APIs where caching causes duplicate records).
 642        """
 643        parent_streams = ConcurrentDeclarativeSource._parent_stream_names(stream_configs)
 644        parent_streams.update(
 645            ConcurrentDeclarativeSource._parent_stream_names(all_stream_configs or [])
 646        )
 647
 648        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
 649            """Set use_cache to True only if not explicitly disabled."""
 650            if requester.get("use_cache") is not False:
 651                requester["use_cache"] = True
 652
 653        def update_with_cache_parent_configs(
 654            parent_configs: list[dict[str, Any]],
 655        ) -> None:
 656            for parent_config in parent_configs:
 657                if parent_config["stream"]["type"] == "StateDelegatingStream":
 658                    _set_cache_if_not_disabled(
 659                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
 660                    )
 661                    _set_cache_if_not_disabled(
 662                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
 663                    )
 664                else:
 665                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
 666
 667        for stream_config in stream_configs:
 668            if stream_config.get("incremental_sync", {}).get("parent_stream"):
 669                _set_cache_if_not_disabled(
 670                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
 671                )
 672
 673            elif stream_config.get("retriever", {}).get("partition_router", {}):
 674                partition_router = stream_config["retriever"]["partition_router"]
 675
 676                routers = (
 677                    list(partition_router)
 678                    if isinstance(partition_router, list)
 679                    else [partition_router]
 680                )
 681                while routers:
 682                    router = routers.pop()
 683                    if not isinstance(router, dict):
 684                        continue
 685                    if router.get("parent_stream_configs"):
 686                        update_with_cache_parent_configs(router["parent_stream_configs"])
 687                    # Descend into composed partition routers (e.g. UnionPartitionRouter's
 688                    # partition_routers or GroupingPartitionRouter's underlying_partition_router)
 689                    # so nested parent streams also get caching enabled.
 690                    routers.extend(router.get("partition_routers") or [])
 691                    if router.get("underlying_partition_router"):
 692                        routers.append(router["underlying_partition_router"])
 693
 694        for stream_config in stream_configs:
 695            if stream_config["name"] in parent_streams:
 696                if stream_config["type"] == "StateDelegatingStream":
 697                    _set_cache_if_not_disabled(
 698                        stream_config["full_refresh_stream"]["retriever"]["requester"]
 699                    )
 700                    _set_cache_if_not_disabled(
 701                        stream_config["incremental_stream"]["retriever"]["requester"]
 702                    )
 703                else:
 704                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
 705        return stream_configs
 706
 707    def spec(self, logger: logging.Logger) -> ConnectorSpecification:
 708        """
 709        Returns the connector specification (spec) as defined in the Airbyte Protocol. The spec is an object describing the possible
 710        configurations (e.g: username and password) which can be configured when running this connector. For low-code connectors, this
 711        will first attempt to load the spec from the manifest's spec block, otherwise it will load it from "spec.yaml" or "spec.json"
 712        in the project root.
 713        """
 714        return (
 715            self._spec_component.generate_spec() if self._spec_component else super().spec(logger)
 716        )
 717
 718    def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
 719        check = self._source_config.get("check")
 720        if not check:
 721            raise ValueError(f"Missing 'check' component definition within the manifest.")
 722
 723        if "type" not in check:
 724            check["type"] = "CheckStream"
 725        connection_checker = self._constructor.create_component(
 726            COMPONENTS_CHECKER_TYPE_MAPPING[check["type"]],
 727            check,
 728            dict(),
 729            emit_connector_builder_messages=self._emit_connector_builder_messages,
 730        )
 731        if not isinstance(connection_checker, ConnectionChecker):
 732            raise ValueError(
 733                f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}"
 734            )
 735
 736        with self._config_overridden_for_check(check.get("config_overrides")):
 737            check_succeeded, error = connection_checker.check_connection(self, logger, self._config)
 738        if not check_succeeded:
 739            return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error))
 740        return AirbyteConnectionStatus(status=Status.SUCCEEDED)
 741
 742    def check_section(
 743        self, logger: logging.Logger, section: str, config: Mapping[str, Any]
 744    ) -> AirbyteConnectionStatus:
 745        """
 746        Section check for the stepped setup form, see `airbyte_cdk.sources.utils.check_section`.
 747
 748        A declarative source validates a configuration by reading from the streams its `check` component names, so a
 749        section check runs that same check on the configuration collected so far. The manifest is responsible for
 750        ordering its `sections` so that the sections which declare `"check": true` already hold every field those
 751        streams need; the entrypoint has validated that the required fields of the section and of the sections before
 752        it are present.
 753        """
 754        logger.info(f"Checking configuration section '{section}' with the declared check streams.")
 755        return self.check(logger, config)
 756
 757    @contextmanager
 758    def _config_overridden_for_check(
 759        self, config_overrides: Optional[Mapping[str, Any]]
 760    ) -> Iterator[None]:
 761        """Overlay the check component's `config_overrides` onto the config for the duration of a check.
 762
 763        Rebinding `self._config` reaches the whole component tree, because `streams()` interpolates from
 764        it and ignores its own `config` argument. Not thread safe; `check` is one command per process.
 765
 766        Semantics:
 767          - values are applied verbatim, never interpolated;
 768          - the merge is one level deep, so an object-valued override replaces rather than merges;
 769          - it happens after `_migrate_and_transform_config`, so derived fields are not recomputed;
 770          - `config_validations` run against `self._config_for_validation`, not the overlay.
 771
 772        The copy is shallow, so a component writing into a nested path writes through to the config the
 773        source was constructed with and the restore does not undo it. Only top-level writes are
 774        discarded, which is why `_raise_if_config_is_persisted` refuses config-persisting manifests.
 775
 776        Known limitation: `$parameters` declared on a check component still propagate into object-valued
 777        overrides. See https://github.com/airbytehq/airbyte-internal-issues/issues/16995.
 778        """
 779        if not config_overrides:
 780            yield
 781            return
 782
 783        self._raise_on_non_string_override_keys(config_overrides)
 784        self._raise_on_reserved_override_keys(config_overrides)
 785        self._raise_if_config_is_persisted(config_overrides)
 786        self._warn_on_unknown_override_keys(config_overrides)
 787        self._register_override_secrets(config_overrides)
 788        # Keys only. An override may name a secret field, so values must not reach the logs.
 789        self.logger.info(
 790            f"Overriding config keys for the check operation: {', '.join(sorted(config_overrides))}"
 791        )
 792
 793        unmodified_config = self._config
 794        self._config = {**self._config, **config_overrides}
 795        try:
 796            yield
 797        finally:
 798            self._config = unmodified_config
 799
 800    @staticmethod
 801    def _raise_on_reserved_override_keys(config_overrides: Mapping[str, Any]) -> None:
 802        """Refuse keys in the platform's reserved `__airbyte` namespace.
 803
 804        `CheckStream` reads `__airbyte_check_stream_names` out of the config this overlay writes to, so
 805        an override there would change which streams a check tests. The whole prefix is refused.
 806        """
 807        reserved = sorted(key for key in config_overrides if key.startswith("__airbyte"))
 808        if reserved:
 809            raise AirbyteTracedException(
 810                message=(
 811                    f"This connector's manifest is invalid: its check component overrides the "
 812                    f"reserved config key(s) {reserved}. Keys prefixed with `__airbyte` belong to "
 813                    "the platform, not to the connector's spec. This is a bug in the connector "
 814                    "rather than in this connection's settings."
 815                ),
 816                internal_message=(
 817                    f"config_overrides rejected: reserved __airbyte key(s) {reserved}. "
 818                    "`CheckStream` reads `__airbyte_check_stream_names` out of the config this "
 819                    "overlay writes to, so an override there would change which streams a check "
 820                    "tests. Use `stream_names` instead."
 821                ),
 822                failure_type=FailureType.system_error,
 823            )
 824
 825    @staticmethod
 826    def _raise_on_non_string_override_keys(config_overrides: Mapping[str, Any]) -> None:
 827        """Reject non-string keys, which YAML allows and every consumer below assumes away."""
 828        non_strings = [key for key in config_overrides if not isinstance(key, str)]
 829        if non_strings:
 830            raise AirbyteTracedException(
 831                message=(
 832                    "This connector's manifest is invalid: its check component has "
 833                    f"`config_overrides` key(s) {sorted(map(repr, non_strings))} that are not "
 834                    "strings, so they cannot name a field in the connector's spec. This is a bug "
 835                    "in the connector rather than in this connection's settings."
 836                ),
 837                internal_message=(
 838                    "config_overrides rejected: non-string key(s) "
 839                    f"{sorted(map(repr, non_strings))}. Quote them in the manifest so they name a "
 840                    "field in the connector's spec."
 841                ),
 842                failure_type=FailureType.system_error,
 843            )
 844
 845    def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> None:
 846        """Reject `config_overrides` on a manifest whose authenticator writes the config back.
 847
 848        A `refresh_token_updater` builds a `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, whose
 849        `_emit_control_message` emits the entire config it was handed as a `CONNECTOR_CONFIG` message.
 850        During a check that is the overlay, and the platform persists it, so a check-only override would
 851        become the connection's saved config. The restore cannot recall a message already on stdout.
 852        """
 853        if not self._manifest_writes_back_config(self._source_config):
 854            return
 855        raise AirbyteTracedException(
 856            message=(
 857                "This connector's manifest is invalid: `config_overrides` cannot be used by a "
 858                "manifest that declares a `refresh_token_updater`. A token refresh during `check` "
 859                "emits the whole config it was handed as a CONNECTOR_CONFIG control message, which "
 860                f"the platform persists, so the check-only override(s) {sorted(config_overrides)} "
 861                "would be saved as this connection's config and applied to every later sync. This "
 862                "is a bug in the connector rather than in this connection's settings."
 863            ),
 864            internal_message=(
 865                "config_overrides rejected: manifest persists config via refresh_token_updater. "
 866                "Remove `config_overrides` from the check component, or drop the "
 867                "`refresh_token_updater`."
 868            ),
 869            failure_type=FailureType.system_error,
 870        )
 871
 872    @staticmethod
 873    def _manifest_writes_back_config(definition: Any) -> bool:
 874        """Whether any component in the manifest emits the connector config back to the platform.
 875
 876        Walks the whole manifest, since an authenticator can sit under any requester, inside a
 877        `SelectiveAuthenticator`, or in a `ConditionalStreams` branch. Callers pass `self._source_config`,
 878        which is post-`_pre_process_manifest`, so a `$ref`-ed authenticator is already inlined and
 879        carries its own `type`.
 880
 881        `is not None` rather than truthiness because `refresh_token_updater: {}` takes every default and
 882        still builds a single-use authenticator - the factory tests a model instance, which is truthy.
 883        The type condition is what separates a component from a data blob that happens to use the name.
 884
 885        Does not detect a `CustomAuthenticator` that persists config without declaring the field.
 886        """
 887        if isinstance(definition, Mapping):
 888            if (
 889                definition.get("refresh_token_updater") is not None
 890                and definition.get("type")
 891                in ConcurrentDeclarativeSource._CONFIG_PERSISTING_AUTHENTICATOR_TYPES
 892            ):
 893                return True
 894            return any(
 895                ConcurrentDeclarativeSource._manifest_writes_back_config(value)
 896                for value in definition.values()
 897            )
 898        if isinstance(definition, list):
 899            return any(
 900                ConcurrentDeclarativeSource._manifest_writes_back_config(item)
 901                for item in definition
 902            )
 903        return False
 904
 905    def _warn_on_unknown_override_keys(self, config_overrides: Mapping[str, Any]) -> None:
 906        """Warn about override keys the spec does not declare.
 907
 908        Nothing validates the overlay - the entrypoint validates what the user supplied, and
 909        `config_validations` run against `self._config_for_validation` - so a typo is a silent no-op.
 910        """
 911        if not self._spec_component:
 912            return
 913        declared = self._declared_config_properties(self._spec_component.connection_specification)
 914        if declared is None:
 915            return
 916        unknown = sorted(key for key in config_overrides if key not in declared)
 917        if unknown:
 918            self.logger.warning(
 919                f"Check-only config override(s) {unknown} are not declared in the connector spec, so "
 920                "they will have no effect on any component that reads the config by field name."
 921            )
 922
 923    @staticmethod
 924    def _declared_config_properties(
 925        connection_specification: Mapping[str, Any],
 926    ) -> Optional[Set[str]]:
 927        """Every field name a spec declares, or `None` when it does not enumerate them.
 928
 929        `oneOf` and `allOf` composition put declarations one level down, so reading only the top-level
 930        `properties` would report a declared field as undeclared.
 931        """
 932        if not isinstance(connection_specification, Mapping):
 933            return None
 934
 935        names: Set[str] = set()
 936        found_any = False
 937        properties = connection_specification.get("properties")
 938        if isinstance(properties, Mapping):
 939            found_any = True
 940            names.update(str(key) for key in properties)
 941        for keyword in ("allOf", "anyOf", "oneOf"):
 942            for branch in connection_specification.get(keyword) or []:
 943                nested = ConcurrentDeclarativeSource._declared_config_properties(branch)
 944                if nested is not None:
 945                    found_any = True
 946                    names.update(nested)
 947        return names if found_any else None
 948
 949    def _register_override_secrets(self, config_overrides: Mapping[str, Any]) -> None:
 950        """Register override values landing on an `airbyte_secret` field, so they get redacted.
 951
 952        The entrypoint builds the secret list from the config the user supplied, so a value substituted
 953        here is unknown to `filter_secrets`. Uses the same discovery as the entrypoint, so the overlay is
 954        redacted exactly where the user's own value at that path would be.
 955        """
 956        if not self._spec_component:
 957            return
 958        for secret in get_secrets(
 959            self._spec_component.connection_specification, dict(config_overrides)
 960        ):
 961            if secret is not None:
 962                add_to_secrets(str(secret))
 963
 964    @property
 965    def dynamic_streams(self) -> List[Dict[str, Any]]:
 966        return self._dynamic_stream_configs(
 967            manifest=self._source_config,
 968            with_dynamic_stream_name=True,
 969        )
 970
 971    def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
 972        # This has a warning flag for static, but after we finish part 4 we'll replace manifest with self._source_config
 973        stream_configs = []
 974        for current_stream_config in manifest.get("streams", []):
 975            if (
 976                "type" in current_stream_config
 977                and current_stream_config["type"] == "ConditionalStreams"
 978            ):
 979                interpolated_boolean = InterpolatedBoolean(
 980                    condition=current_stream_config.get("condition"),
 981                    parameters={},
 982                )
 983
 984                if interpolated_boolean.eval(config=self._config):
 985                    stream_configs.extend(current_stream_config.get("streams", []))
 986            else:
 987                if "type" not in current_stream_config:
 988                    current_stream_config["type"] = "DeclarativeStream"
 989                stream_configs.append(current_stream_config)
 990        return stream_configs
 991
 992    def _dynamic_stream_configs(
 993        self,
 994        manifest: Mapping[str, Any],
 995        with_dynamic_stream_name: Optional[bool] = None,
 996    ) -> List[Dict[str, Any]]:
 997        dynamic_stream_definitions: List[Dict[str, Any]] = manifest.get("dynamic_streams", [])
 998        dynamic_stream_configs: List[Dict[str, Any]] = []
 999        seen_dynamic_streams: Set[str] = set()
1000
1001        for dynamic_definition_index, dynamic_definition in enumerate(dynamic_stream_definitions):
1002            components_resolver_config = dynamic_definition["components_resolver"]
1003
1004            if not components_resolver_config:
1005                raise ValueError(
1006                    f"Missing 'components_resolver' in dynamic definition: {dynamic_definition}"
1007                )
1008
1009            resolver_type = components_resolver_config.get("type")
1010            if not resolver_type:
1011                raise ValueError(
1012                    f"Missing 'type' in components resolver configuration: {components_resolver_config}"
1013                )
1014
1015            if resolver_type not in COMPONENTS_RESOLVER_TYPE_MAPPING:
1016                raise ValueError(
1017                    f"Invalid components resolver type '{resolver_type}'. "
1018                    f"Expected one of {list(COMPONENTS_RESOLVER_TYPE_MAPPING.keys())}."
1019                )
1020
1021            # AsyncRetriever has no `requester`, and a cached polling request replays a stale status
1022            if "requester" in components_resolver_config.get("retriever", {}):
1023                components_resolver_config["retriever"]["requester"]["use_cache"] = True
1024
1025            # Create a resolver for dynamic components based on type
1026            if resolver_type == "HttpComponentsResolver":
1027                components_resolver = self._constructor.create_component(
1028                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
1029                    component_definition=components_resolver_config,
1030                    config=self._config,
1031                    stream_name=dynamic_definition.get("name"),
1032                )
1033            else:
1034                components_resolver = self._constructor.create_component(
1035                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
1036                    component_definition=components_resolver_config,
1037                    config=self._config,
1038                )
1039
1040            stream_template_config = dynamic_definition["stream_template"]
1041
1042            for dynamic_stream in components_resolver.resolve_components(
1043                stream_template_config=stream_template_config
1044            ):
1045                # Get the use_parent_parameters configuration from the dynamic definition
1046                # Default to True for backward compatibility, since connectors were already using it by default when this param was added
1047                use_parent_parameters = dynamic_definition.get("use_parent_parameters", True)
1048
1049                dynamic_stream = {
1050                    **ManifestComponentTransformer().propagate_types_and_parameters(
1051                        "", dynamic_stream, {}, use_parent_parameters=use_parent_parameters
1052                    )
1053                }
1054
1055                if "type" not in dynamic_stream:
1056                    dynamic_stream["type"] = "DeclarativeStream"
1057
1058                # Ensure that each stream is created with a unique name
1059                name = dynamic_stream.get("name")
1060
1061                if with_dynamic_stream_name:
1062                    dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
1063                        "name", f"dynamic_stream_{dynamic_definition_index}"
1064                    )
1065
1066                if not isinstance(name, str):
1067                    raise ValueError(
1068                        f"Expected stream name {name} to be a string, got {type(name)}."
1069                    )
1070
1071                if name in seen_dynamic_streams:
1072                    error_message = f"Dynamic streams list contains a duplicate name: {name}. Please contact Airbyte Support."
1073                    failure_type = FailureType.system_error
1074
1075                    if resolver_type == "ConfigComponentsResolver":
1076                        error_message = f"Dynamic streams list contains a duplicate name: {name}. Please check your configuration."
1077                        failure_type = FailureType.config_error
1078
1079                    raise AirbyteTracedException(
1080                        message=error_message,
1081                        internal_message=error_message,
1082                        failure_type=failure_type,
1083                    )
1084
1085                seen_dynamic_streams.add(name)
1086                dynamic_stream_configs.append(dynamic_stream)
1087
1088        return dynamic_stream_configs
1089
1090    def _select_streams(
1091        self, streams: List[AbstractStream], configured_catalog: ConfiguredAirbyteCatalog
1092    ) -> List[AbstractStream]:
1093        stream_name_to_instance: Mapping[str, AbstractStream] = {s.name: s for s in streams}
1094        abstract_streams: List[AbstractStream] = []
1095        for configured_stream in configured_catalog.streams:
1096            stream_instance = stream_name_to_instance.get(configured_stream.stream.name)
1097            if stream_instance:
1098                abstract_streams.append(stream_instance)
1099            else:
1100                # Previous behavior in the legacy synchronous CDK was to also raise an error TRACE message if
1101                # the source was configured with raise_exception_on_missing_stream=True. This was used on very
1102                # few sources like facebook-marketing and google-ads. We decided not to port this feature over,
1103                # but we can do so if we feel it necessary. With the current behavior,we should still result
1104                # in a partial failure since missing streams will be marked as INCOMPLETE.
1105                self._message_repository.emit_message(
1106                    as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE)
1107                )
1108        return abstract_streams