fix(low-code): evaluate concurrency_level with the normalized config

[Baz (bazarnov)]

Overview

👉 TL;DR: Connectors now size their worker pool from the config after its migrations and transformations run, not from the config as stored. Before, a migration that changed the worker count took effect one sync late, and a transformation that changed it never took effect. Specifically, ConcurrentDeclarativeSource.__init__ builds the ConcurrencyLevel component from self._config instead of the raw config argument, so default_concurrency interpolates the output of spec.config_normalization_rules.

Changes

• concurrent_declarative_source.py: pass config=self._config to the ConcurrencyLevel component instead of config=config or {}. Streams, authenticators, api_budget, dynamic streams and check already read self._config.
• test_concurrent_declarative_source.py: add test_concurrency_level_uses_normalized_config, parametrized over config_migrations and transformations. Each case rewrites num_workers from 1 to 4 and asserts the thread pool has 4 workers.

Review Spotlight

Reviewers with limited time, please review first:

• concurrent_declarative_source.py#L239-L251: self._config is never None (or {}), so it replaces config or {} directly
• concurrent_declarative_source.py#L355-L373: without a spec, self._config is the raw config, so those manifests see no change
• test_concurrent_declarative_source.py#L1800-L1833: the new test

Risks

• Behaviour change: the pool size changes only when a config migration or transformation rewrites a field that default_concurrency reads. The platform saves migrated configs, so a migration was missed on one run per connection. Transformations are never saved, so one that drives default_concurrency was ignored on every run.
• Who sees it: 2 monorepo connectors.
  • source-stripe raises num_workers from 1 to 2 in a migration. On the run that migrates a stored num_workers: 1, the pool now has 2 workers instead of 1, the same as later runs. Built from the real manifest with a dummy config: pool size 1 on main, 2 with the fix.
  • source-zendesk-support raises num_workers below 2 to 2, and its [config.get('num_workers', 4), 2] | max clamp already yields 2, so its pool size is unchanged. After it moves to a CDK release with this fix, the clamp becomes optional. No action required.
• How it was counted (airbyte master): 537 connector manifest.yaml files (git ls-tree -r --name-only origin/master -- airbyte-integrations/connectors | grep -E '/manifest\.yaml$'); 54 declare concurrency_level (git grep -l concurrency_level origin/master -- 'airbyte-integrations/connectors/*/manifest.yaml' 'airbyte-integrations/connectors/*/**/manifest.yaml'), 39 of them interpolate config and 15 use a constant; the same grep finds 20 with config_normalization_rules, all of which also declare concurrency_level. For those 20, a script compared the config fields each default_concurrency reads with the fields each migration or transformation writes. The custom transformations in source-github, source-mailchimp and source-google-analytics-data-api were read by hand and do not affect concurrency.
• Not affected: manifests without config_normalization_rules, the Connector Builder and the manifest server (both force default_concurrency: 1). Records, state, catalogs and the manifest schema are unchanged. Custom manifests in Cloud cannot be scanned; their concurrency now follows the same config their streams already read.

Test plan

• python -m pytest "unit_tests/sources/declarative/test_concurrent_declarative_source.py::test_concurrency_level_uses_normalized_config" -q -p no:cacheprovider
  • main source (git checkout origin/main -- airbyte_cdk/sources/declarative/concurrent_declarative_source.py): 2 failed, assert 1 == 4
  • fix: 2 passed
• python -m pytest unit_tests/sources/declarative/test_concurrent_declarative_source.py unit_tests/sources/declarative/parsers unit_tests/legacy -q -p no:cacheprovider -k "not test_read_with_concurrent_and_synchronous_streams": 647 passed, 3 deselected. The 3 deselected tests fail locally on macOS with the SQLite cache race from Cached parent stream with more than one partition corrupts the shared sqlite request cache under concurrent reads (no such table: responses, file is not a database, no more rows available) #1150, on main too.
• python -m pytest unit_tests/connector_builder -q -p no:cacheprovider: 63 passed.
• ruff format --check and ruff check on both files, mypy --config-file mypy.ini airbyte_cdk/sources/declarative/concurrent_declarative_source.py: clean.

Fixes #1147

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@bazarnov/fix-concurrency-level-normalized-config#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch bazarnov/fix-concurrency-level-normalized-config

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 984 tests  +2   4 972 ✅ +2   8m 58s ⏱️ -35s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 146d5ce. ± Comparison against base commit d214b0c.

[github-actions]

PyTest Results (Full)

4 987 tests  +2   4 975 ✅ +2   16m 4s ⏱️ ±0s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit 146d5ce. ± Comparison against base commit d214b0c.