Merge branch 'main' into devin/1754519222-add-json-schema-generation

Author: aaronsteers

.github/actions/check-docker-tag/action.yml

@@ -0,0 +1,28 @@
+name: 'Check Docker Tag Exists'
+description: 'Check if a Docker tag exists on DockerHub to prevent overwrites'
+inputs:
+  image_name:
+    description: 'Docker image name (e.g. airbyte/source-declarative-manifest)'
+    required: true
+  tag:
+    description: 'Docker tag to check'
+    required: true
+runs:
+  using: "composite"
+  steps:
+    - name: "Check for existing tag (${{ inputs.image_name }}:${{ inputs.tag }})"
+      shell: bash
+      run: |
+        image="${{ inputs.image_name }}"
+        tag_input="${{ inputs.tag }}"
+        if [ -z "$image" ] || [ -z "$tag_input" ]; then
+          echo "Error: image_name and tag are required."
+          exit 1
+        fi
+        tag="${image}:${tag_input}"
+        echo "Checking if tag '$tag' exists on DockerHub..."
+        if DOCKER_CLI_EXPERIMENTAL=enabled docker manifest inspect "$tag" > /dev/null 2>&1; then
+          echo "The tag '$tag' already exists on DockerHub. Skipping publish to prevent overwrite."
+          exit 1
+        fi
+        echo "No existing tag '$tag' found. Proceeding with publish."

.github/workflows/docker-build-check.yml

@@ -1,13 +1,15 @@
 name: Docker Build Check
+permissions:
+  contents: read
 
 on:
   pull_request:
     branches:
       - main
 
 jobs:
-  docker-build-check:
-    name: SDM Docker Image Build # Renamed job to be more descriptive
+  sdm-docker-build-check:
+    name: SDM Docker Image Build
     runs-on: ubuntu-24.04
     steps:
       - name: Checkout code
@@ -42,3 +44,29 @@ jobs:
           push: false
           tags: airbyte/source-declarative-manifest:pr-${{ github.event.pull_request.number }}
           outputs: type=image,name=target,annotation-index.org.opencontainers.image.description=SDM Docker image for PR ${{ github.event.pull_request.number }}
+
+  manifest-server-docker-build-check:
+    name: Manifest Server Docker Image Build
+    runs-on: ubuntu-24.04
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Set up QEMU for multi-platform builds
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build Manifest Server Docker image for multiple platforms
+        id: manifest-server-build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: false
+          tags: airbyte/manifest-server:pr-${{ github.event.pull_request.number }}
+          outputs: type=image,name=target,annotation-index.org.opencontainers.image.description=Manifest Server Docker image for PR ${{ github.event.pull_request.number }}

.github/workflows/pypi_publish.yml .github/workflows/publish.yml.github/workflows/pypi_publish.yml renamed to .github/workflows/publish.yml

@@ -6,6 +6,8 @@
 # we have to also update the Trusted Publisher settings on PyPI.
 
 name: CDK Publish
+permissions:
+  contents: read
 
 on:
   push:
@@ -31,6 +33,11 @@ on:
         type: boolean
         required: true
         default: true
+      publish_manifest_server:
+        description: "Publish Manifest Server to DockerHub. If true, the workflow will publish the Manifest Server to DockerHub."
+        type: boolean
+        required: true
+        default: true
       update_connector_builder:
         description: "Update Connector Builder. If true, the workflow will create a PR to bump the CDK version used by Connector Builder."
         type: boolean
@@ -204,18 +211,10 @@ jobs:
 
       - name: "Check for existing tag (version: ${{ env.VERSION || 'none' }} )"
         if: env.VERSION != ''
-        run: |
-          tag="airbyte/source-declarative-manifest:${{ env.VERSION }}"
-          if [ -z "$tag" ]; then
-            echo "Error: VERSION is not set. Ensure the tag follows the format 'refs/tags/vX.Y.Z'."
-            exit 1
-          fi
-          echo "Checking if tag '$tag' exists on DockerHub..."
-          if DOCKER_CLI_EXPERIMENTAL=enabled docker manifest inspect "$tag" > /dev/null 2>&1; then
-            echo "The tag '$tag' already exists on DockerHub. Skipping publish to prevent overwrite."
-            exit 1
-          fi
-          echo "No existing tag '$tag' found. Proceeding with publish."
+        uses: ./.github/actions/check-docker-tag
+        with:
+          image_name: airbyte/source-declarative-manifest
+          tag: ${{ env.VERSION }}
 
       - name: "Build and push (sha tag: '${{ github.sha }}')"
         # Only run if the version is not set
@@ -250,6 +249,90 @@ jobs:
           tags: |
             airbyte/source-declarative-manifest:latest
 
+  publish_manifest_server:
+    name: Publish Manifest Server to DockerHub
+    if: >
+      (github.event_name == 'push' &&
+       startsWith(github.ref, 'refs/tags/v')) ||
+      (github.event_name == 'workflow_dispatch' &&
+       github.event.inputs.publish_manifest_server == 'true'
+      )
+    runs-on: ubuntu-24.04
+    needs: [build]
+    environment:
+      name: DockerHub
+      url: https://hub.docker.com/r/airbyte/manifest-server/tags
+    env:
+      VERSION: ${{ needs.build.outputs.VERSION }}
+      IS_PRERELEASE: ${{ needs.build.outputs.IS_PRERELEASE }}
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      # We need to download the build artifact again because the previous job was on a different runner
+      - name: Download Build Artifact
+        uses: actions/download-artifact@v4
+        with:
+          name: Packages-${{ github.run_id }}
+          path: dist
+
+      - name: Set up QEMU for multi-platform builds
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Login to Docker Hub
+        uses: docker/login-action@v3
+        with:
+          username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          password: ${{ secrets.DOCKER_HUB_PASSWORD }}
+
+      - name: "Check for existing tag (version: ${{ env.VERSION || 'none' }} )"
+        if: env.VERSION != ''
+        uses: ./.github/actions/check-docker-tag
+        with:
+          image_name: airbyte/manifest-server
+          tag: ${{ env.VERSION }}
+
+      - name: "Build and push (sha tag: '${{ github.sha }}')"
+        # Only run if the version is not set
+        if: env.VERSION == ''
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:${{ github.sha }}
+
+      - name: "Build and push (version tag: ${{ env.VERSION || 'none'}})"
+        # Only run if the version is set
+        if: env.VERSION != ''
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:${{ env.VERSION }}
+
+      - name: Build and push ('latest' tag)
+        # Only run if version is set and IS_PRERELEASE is false
+        if: env.VERSION != '' && env.IS_PRERELEASE == 'false'
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: airbyte_cdk/manifest_server/Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: true
+          tags: |
+            airbyte/manifest-server:latest
+
   update-connector-builder:
     # Create a PR against the Builder, to update the CDK version that it uses.
     # In the future, Builder may use the SDM docker image instead of the Python CDK package.

airbyte_cdk/__init__.py

@@ -107,7 +107,6 @@
 from .sources.declarative.extractors.record_filter import RecordFilter
 from .sources.declarative.incremental import DatetimeBasedCursor
 from .sources.declarative.interpolation import InterpolatedBoolean, InterpolatedString
-from .sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from .sources.declarative.migrations.legacy_to_per_partition_state_migration import (
     LegacyToPerPartitionStateMigration,
 )
@@ -253,7 +252,6 @@
     "JsonDecoder",
     "JsonFileSchemaLoader",
     "LegacyToPerPartitionStateMigration",
-    "ManifestDeclarativeSource",
     "MinMaxDatetime",
     "NoAuth",
     "OffsetIncrement",

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -3,8 +3,8 @@
 #
 
 
-from dataclasses import asdict, dataclass, field
-from typing import Any, ClassVar, Dict, List, Mapping
+from dataclasses import asdict
+from typing import Any, Dict, List, Mapping, Optional
 
 from airbyte_cdk.connector_builder.test_reader import TestReader
 from airbyte_cdk.models import (
@@ -15,45 +15,30 @@
     Type,
 )
 from airbyte_cdk.models import Type as MessageType
-from airbyte_cdk.sources.declarative.declarative_source import DeclarativeSource
-from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
-from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
-    ModelToComponentFactory,
+from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
+    ConcurrentDeclarativeSource,
+    TestLimits,
 )
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
 from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
-DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE = 5
-DEFAULT_MAXIMUM_NUMBER_OF_SLICES = 5
-DEFAULT_MAXIMUM_RECORDS = 100
-DEFAULT_MAXIMUM_STREAMS = 100
-
 MAX_PAGES_PER_SLICE_KEY = "max_pages_per_slice"
 MAX_SLICES_KEY = "max_slices"
 MAX_RECORDS_KEY = "max_records"
 MAX_STREAMS_KEY = "max_streams"
 
 
-@dataclass
-class TestLimits:
-    __test__: ClassVar[bool] = False  # Tell Pytest this is not a Pytest class, despite its name
-
-    max_records: int = field(default=DEFAULT_MAXIMUM_RECORDS)
-    max_pages_per_slice: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE)
-    max_slices: int = field(default=DEFAULT_MAXIMUM_NUMBER_OF_SLICES)
-    max_streams: int = field(default=DEFAULT_MAXIMUM_STREAMS)
-
-
 def get_limits(config: Mapping[str, Any]) -> TestLimits:
     command_config = config.get("__test_read_config", {})
-    max_pages_per_slice = (
-        command_config.get(MAX_PAGES_PER_SLICE_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE
+    return TestLimits(
+        max_records=command_config.get(MAX_RECORDS_KEY, TestLimits.DEFAULT_MAX_RECORDS),
+        max_pages_per_slice=command_config.get(
+            MAX_PAGES_PER_SLICE_KEY, TestLimits.DEFAULT_MAX_PAGES_PER_SLICE
+        ),
+        max_slices=command_config.get(MAX_SLICES_KEY, TestLimits.DEFAULT_MAX_SLICES),
+        max_streams=command_config.get(MAX_STREAMS_KEY, TestLimits.DEFAULT_MAX_STREAMS),
     )
-    max_slices = command_config.get(MAX_SLICES_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_SLICES
-    max_records = command_config.get(MAX_RECORDS_KEY) or DEFAULT_MAXIMUM_RECORDS
-    max_streams = command_config.get(MAX_STREAMS_KEY) or DEFAULT_MAXIMUM_STREAMS
-    return TestLimits(max_records, max_pages_per_slice, max_slices, max_streams)
 
 
 def should_migrate_manifest(config: Mapping[str, Any]) -> bool:
@@ -75,26 +60,35 @@ def should_normalize_manifest(config: Mapping[str, Any]) -> bool:
     return config.get("__should_normalize", False)
 
 
-def create_source(config: Mapping[str, Any], limits: TestLimits) -> ManifestDeclarativeSource:
+def create_source(
+    config: Mapping[str, Any],
+    limits: TestLimits,
+    catalog: Optional[ConfiguredAirbyteCatalog],
+    state: Optional[List[AirbyteStateMessage]],
+) -> ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]]:
     manifest = config["__injected_declarative_manifest"]
-    return ManifestDeclarativeSource(
+
+    # We enforce a concurrency level of 1 so that the stream is processed on a single thread
+    # to retain ordering for the grouping of the builder message responses.
+    if "concurrency_level" in manifest:
+        manifest["concurrency_level"]["default_concurrency"] = 1
+    else:
+        manifest["concurrency_level"] = {"type": "ConcurrencyLevel", "default_concurrency": 1}
+
+    return ConcurrentDeclarativeSource(
+        catalog=catalog,
         config=config,
-        emit_connector_builder_messages=True,
+        state=state,
         source_config=manifest,
+        emit_connector_builder_messages=True,
         migrate_manifest=should_migrate_manifest(config),
         normalize_manifest=should_normalize_manifest(config),
-        component_factory=ModelToComponentFactory(
-            emit_connector_builder_messages=True,
-            limit_pages_fetched_per_slice=limits.max_pages_per_slice,
-            limit_slices_fetched=limits.max_slices,
-            disable_retries=True,
-            disable_cache=True,
-        ),
+        limits=limits,
     )
 
 
 def read_stream(
-    source: DeclarativeSource,
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
     config: Mapping[str, Any],
     configured_catalog: ConfiguredAirbyteCatalog,
     state: List[AirbyteStateMessage],
@@ -132,7 +126,9 @@ def read_stream(
         return error.as_airbyte_message()
 
 
-def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
+def resolve_manifest(
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]],
+) -> AirbyteMessage:
     try:
         return AirbyteMessage(
             type=Type.RECORD,
@@ -149,7 +145,9 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
         return error.as_airbyte_message()
 
 
-def full_resolve_manifest(source: ManifestDeclarativeSource, limits: TestLimits) -> AirbyteMessage:
+def full_resolve_manifest(
+    source: ConcurrentDeclarativeSource[Optional[List[AirbyteStateMessage]]], limits: TestLimits
+) -> AirbyteMessage:
     try:
         manifest = {**source.resolved_manifest}
         streams = manifest.get("streams", [])