only apply concurrency changes on the test reads

Author: maxime.c

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:

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,