feat: enhance concurrency logging to differentiate config vs default values

Co-Authored-By: unknown <>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -233,12 +233,23 @@ def __init__(
             concurrency_level = self._LOWEST_SAFE_CONCURRENCY_LEVEL
             initial_number_of_partitions_to_generate = self._LOWEST_SAFE_CONCURRENCY_LEVEL // 2
 
-        self.logger.info(
-            "Concurrency configuration: concurrency_level=%d, initial_number_of_partitions_to_generate=%d, concurrency_level_from_manifest=%s",
-            concurrency_level,
-            initial_number_of_partitions_to_generate,
-            "defined" if concurrency_level_from_manifest else "not_defined",
-        )
+        if concurrency_level_from_manifest:
+            raw_default_concurrency = concurrency_level_from_manifest.get("default_concurrency", "N/A")
+            self.logger.info(
+                "Concurrency configuration: concurrency_level=%d, initial_number_of_partitions_to_generate=%d, "
+                "source=manifest (expression=%s), config=%s",
+                concurrency_level,
+                initial_number_of_partitions_to_generate,
+                raw_default_concurrency,
+                {k: v for k, v in (config or {}).items() if "worker" in k.lower() or "concurren" in k.lower()},
+            )
+        else:
+            self.logger.info(
+                "Concurrency configuration: concurrency_level=%d, initial_number_of_partitions_to_generate=%d, "
+                "source=default (_LOWEST_SAFE_CONCURRENCY_LEVEL)",
+                concurrency_level,
+                initial_number_of_partitions_to_generate,
+            )
 
         self._concurrent_source = ConcurrentSource.create(
             num_workers=concurrency_level,