diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
index 0c5e9da08c..055576f45f 100644
--- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
+++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py
@@ -247,7 +247,7 @@ def __init__(
             concurrency_level_component = self._constructor.create_component(
                 model_type=ConcurrencyLevelModel,
                 component_definition=concurrency_level_from_manifest,
-                config=config or {},
+                config=self._config,
             )
             if not isinstance(concurrency_level_component, ConcurrencyLevel):
                 raise ValueError(
diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py
index 47b80df66b..c25831b9ab 100644
--- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py
+++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py
@@ -1797,6 +1797,42 @@ def test_concurrency_level_initial_number_partitions_to_generate_is_always_one_o
     assert source._concurrent_source._initial_number_partitions_to_generate == 1
 
 
+_SET_NUM_WORKERS_TO_4 = {
+    "type": "ConfigAddFields",
+    "fields": [{"type": "AddedFieldDefinition", "path": ["num_workers"], "value": "4"}],
+}
+
+
+@pytest.mark.parametrize(
+    "config_normalization_rules",
+    [
+        pytest.param(
+            {
+                "config_migrations": [
+                    {"type": "ConfigMigration", "transformations": [_SET_NUM_WORKERS_TO_4]}
+                ]
+            },
+            id="config_migrations",
+        ),
+        pytest.param({"transformations": [_SET_NUM_WORKERS_TO_4]}, id="transformations"),
+    ],
+)
+def test_concurrency_level_uses_normalized_config(config_normalization_rules):
+    manifest = copy.deepcopy(_MANIFEST)
+    manifest["spec"] = {
+        "type": "Spec",
+        "connection_specification": {},
+        "config_normalization_rules": {
+            "type": "ConfigNormalizationRules",
+            **config_normalization_rules,
+        },
+    }
+
+    source = ConcurrentDeclarativeSource(source_config=manifest, config={"num_workers": 1})
+
+    assert source._concurrent_source._threadpool._threadpool._max_workers == 4
+
+
 def test_async_incremental_stream_uses_concurrent_cursor_with_state():
     state = [
         AirbyteStateMessage(