From 3205424d7e22f1a9813da291979ff59a1c6908fd Mon Sep 17 00:00:00 2001 From: "maxime.c" Date: Wed, 15 Oct 2025 14:32:36 -0400 Subject: [PATCH 1/6] fix default concurrency settings --- airbyte_cdk/manifest_server/command_processor/utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airbyte_cdk/manifest_server/command_processor/utils.py b/airbyte_cdk/manifest_server/command_processor/utils.py index 3aef171e06..fc2fc94d55 100644 --- a/airbyte_cdk/manifest_server/command_processor/utils.py +++ b/airbyte_cdk/manifest_server/command_processor/utils.py @@ -72,7 +72,7 @@ def build_source( else: definition["concurrency_level"] = { "type": "ConcurrencyLevel", - "default_concurrency": 1, + "default_concurrency": 2, } should_normalize = should_normalize_manifest(manifest) From db8312d692e2d2ecea9914880106360fea1da17e Mon Sep 17 00:00:00 2001 From: "maxime.c" Date: Wed, 15 Oct 2025 14:55:52 -0400 Subject: [PATCH 2/6] only apply concurrency changes on the test reads --- .../manifest_server/command_processor/utils.py | 9 --------- airbyte_cdk/manifest_server/routers/manifest.py | 14 +++++++++++++- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/airbyte_cdk/manifest_server/command_processor/utils.py b/airbyte_cdk/manifest_server/command_processor/utils.py index fc2fc94d55..2f8b3c30f0 100644 --- a/airbyte_cdk/manifest_server/command_processor/utils.py +++ b/airbyte_cdk/manifest_server/command_processor/utils.py @@ -64,16 +64,7 @@ def build_source( page_limit: Optional[int] = None, slice_limit: Optional[int] = None, ) -> ConcurrentDeclarativeSource: - # 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. definition = copy.deepcopy(manifest) - if "concurrency_level" in definition: - definition["concurrency_level"]["default_concurrency"] = 1 - else: - definition["concurrency_level"] = { - "type": "ConcurrencyLevel", - "default_concurrency": 2, - } should_normalize = should_normalize_manifest(manifest) if should_normalize: diff --git a/airbyte_cdk/manifest_server/routers/manifest.py b/airbyte_cdk/manifest_server/routers/manifest.py index 4fefb21293..d94366e30a 100644 --- a/airbyte_cdk/manifest_server/routers/manifest.py +++ b/airbyte_cdk/manifest_server/routers/manifest.py @@ -1,3 +1,4 @@ +import copy import hashlib from dataclasses import asdict from typing import Any, Dict, List, Mapping, Optional @@ -87,8 +88,19 @@ def test_read(request: StreamTestReadRequest) -> StreamReadResponse: "md5": hashlib.md5(request.custom_components_code.encode()).hexdigest() } + # 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. + manifest = copy.deepcopy(request.manifest.model_dump()) + if "concurrency_level" in manifest: + manifest["concurrency_level"]["default_concurrency"] = 1 + else: + manifest["concurrency_level"] = { + "type": "ConcurrencyLevel", + "default_concurrency": 1, + } + source = safe_build_source( - request.manifest.model_dump(), + manifest, config_dict, catalog, converted_state, From 20cece8e9e6541e8df2a6db87f2d5d06ff0eeb3a Mon Sep 17 00:00:00 2001 From: "maxime.c" Date: Wed, 15 Oct 2025 15:01:47 -0400 Subject: [PATCH 3/6] remove deep copy --- airbyte_cdk/manifest_server/routers/manifest.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/airbyte_cdk/manifest_server/routers/manifest.py b/airbyte_cdk/manifest_server/routers/manifest.py index d94366e30a..0ccf74fb0b 100644 --- a/airbyte_cdk/manifest_server/routers/manifest.py +++ b/airbyte_cdk/manifest_server/routers/manifest.py @@ -1,4 +1,3 @@ -import copy import hashlib from dataclasses import asdict from typing import Any, Dict, List, Mapping, Optional @@ -90,7 +89,7 @@ def test_read(request: StreamTestReadRequest) -> StreamReadResponse: # 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. - manifest = copy.deepcopy(request.manifest.model_dump()) + manifest = request.manifest.model_dump() if "concurrency_level" in manifest: manifest["concurrency_level"]["default_concurrency"] = 1 else: From ea28f544086be27ec81a7738181a41794ad4d1da Mon Sep 17 00:00:00 2001 From: "maxime.c" Date: Wed, 15 Oct 2025 15:10:29 -0400 Subject: [PATCH 4/6] fix tests --- .../command_processor/test_utils.py | 55 ------------------- .../manifest_server/routers/test_manifest.py | 14 ++++- 2 files changed, 12 insertions(+), 57 deletions(-) diff --git a/unit_tests/manifest_server/command_processor/test_utils.py b/unit_tests/manifest_server/command_processor/test_utils.py index 81b43f3035..708bca8811 100644 --- a/unit_tests/manifest_server/command_processor/test_utils.py +++ b/unit_tests/manifest_server/command_processor/test_utils.py @@ -31,61 +31,6 @@ def test_build_catalog_creates_correct_structure(self): assert configured_stream.sync_mode == SyncMode.incremental assert configured_stream.destination_sync_mode == DestinationSyncMode.overwrite - @patch("airbyte_cdk.manifest_server.command_processor.utils.ConcurrentDeclarativeSource") - def test_build_source_creates_source(self, mock_source_class): - """Test that build_source creates a ConcurrentDeclarativeSource with correct parameters.""" - # Setup mocks - mock_source = Mock() - mock_source_class.return_value = mock_source - - # Test with complex manifest and config structures - manifest = { - "version": "0.1.0", - "definitions": {"selector": {"extractor": {"field_path": ["data"]}}}, - "streams": [ - { - "name": "users", - "primary_key": "id", - "retriever": { - "requester": { - "url_base": "https://api.example.com", - "path": "/users", - } - }, - } - ], - "check": {"stream_names": ["users"]}, - } - - config = { - "api_key": "sk-test-123", - "base_url": "https://api.example.com", - "timeout": 30, - } - - # Call build_source with additional parameters - catalog = build_catalog("test_stream") - state = [] - result = build_source(manifest, catalog, config, state) - - # Verify ConcurrentDeclarativeSource was created with correct parameters - expected_source_config = { - **manifest, - "concurrency_level": {"type": "ConcurrencyLevel", "default_concurrency": 1}, - } - mock_source_class.assert_called_once_with( - catalog=catalog, - state=state, - source_config=expected_source_config, - config=config, - normalize_manifest=False, # Default when flag not set - migrate_manifest=False, # Default when flag not set - emit_connector_builder_messages=True, - limits=mock_source_class.call_args[1]["limits"], - ) - - assert result == mock_source - @patch("airbyte_cdk.manifest_server.command_processor.utils.ConcurrentDeclarativeSource") def test_build_source_with_normalize_flag(self, mock_source_class): """Test build_source when normalize flag is set.""" diff --git a/unit_tests/manifest_server/routers/test_manifest.py b/unit_tests/manifest_server/routers/test_manifest.py index 4f8a90b933..5a97a3f16d 100644 --- a/unit_tests/manifest_server/routers/test_manifest.py +++ b/unit_tests/manifest_server/routers/test_manifest.py @@ -109,8 +109,12 @@ def test_test_read_endpoint_success( assert response.status_code == 200 # Verify build_source was called with correct arguments + expected_source_config = { + **sample_manifest, + "concurrency_level": {"type": "ConcurrencyLevel", "default_concurrency": 1}, + } mock_build_source.assert_called_once_with( - sample_manifest, + expected_source_config, mock_build_catalog.return_value, sample_config, [], @@ -168,9 +172,15 @@ def test_test_read_with_custom_components( assert "__injected_components_py_checksums" in config_arg assert config_arg["__injected_components_py_checksums"]["md5"] == expected_checksum + # Verify other arguments + # Verify build_source was called with correct arguments + expected_source_config = { + **sample_manifest, + "concurrency_level": {"type": "ConcurrencyLevel", "default_concurrency": 1}, + } mock_build_source.assert_called_once_with( - sample_manifest, + expected_source_config, mock_build_catalog.return_value, config_arg, [], From dbe666172a1750bad25038e89af61b2370415477 Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Wed, 15 Oct 2025 19:12:24 +0000 Subject: [PATCH 5/6] Auto-fix lint and format issues --- unit_tests/manifest_server/routers/test_manifest.py | 1 - 1 file changed, 1 deletion(-) diff --git a/unit_tests/manifest_server/routers/test_manifest.py b/unit_tests/manifest_server/routers/test_manifest.py index 5a97a3f16d..29940488be 100644 --- a/unit_tests/manifest_server/routers/test_manifest.py +++ b/unit_tests/manifest_server/routers/test_manifest.py @@ -172,7 +172,6 @@ def test_test_read_with_custom_components( assert "__injected_components_py_checksums" in config_arg assert config_arg["__injected_components_py_checksums"]["md5"] == expected_checksum - # Verify other arguments # Verify build_source was called with correct arguments expected_source_config = { From 1991c074e0e5ae1e51968f7d277e25ea3a9df289 Mon Sep 17 00:00:00 2001 From: "maxime.c" Date: Wed, 15 Oct 2025 16:06:47 -0400 Subject: [PATCH 6/6] trigger ci