airbytehq / airbytehq/airbyte-python-cdk
declarative: ConcurrencyLevel is built from the pre-migration config, so config migrations do not affect concurrency on the sync that performs them
- 主要语言
- Python
- 星标
- 26
- 派生
- 53
- 平均合并
- 2 天 6 小时
- 30 天内合并 PR
- 10
描述
## Symptom
A `ConfigMigration` declared under `spec.config_normalization_rules` that changes a field consumed by `concurrency_level.default_concurrency` has no effect on the sync that performs the migration. The migrated value is emitted in the `CONNECTOR_CONFIG` control message and is visible to streams, but the worker pool is sized from the original, unmigrated value. Only the next run, after the platform has persisted the migrated config, picks it up.
## Root cause
`ConcurrentDeclarativeSource.__init__` migrates and transforms the config into `self._config`, then builds the `ConcurrencyLevel` component from the raw `config` argument instead of `self._config`.
- `self._config = self._migrate_and_transform_config(config_path, config) or {}`
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L224
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L239
- `create_component(model_type=ConcurrencyLevelModel, component_definition=concurrency_level_from_manifest, config=config or {})` - note `config`, not `self._config`
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L228-L232 (the `config=config or {}` line is L231)
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L247-L251 (the `config=config or {}` line is L250)
- `_migrate_and_transform_config` copies first (`mutable_config = dict(config)`) and mutates only the copy, so the `config` argument stays pre-migration
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L336-L354 (copy at L345, `migrate_config` at L346, control message at L351, `transform_config` at L353)
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L355-L373 (copy at L364, `migrate_config` at L365, control message at L370, `transform_config` at L372)
- The same applies to `config_normalization_rules.transformations`: `Spec.transform_config` also mutates only the copy, so per-sync config transformations are equally invisible to `ConcurrencyLevel`
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/spec/spec.py#L74-L91 (`migrate_config` L74-L82, `transform_config` L84-L91)
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/spec/spec.py#L74-L91
- `ConcurrencyLevel.get_concurrency_level` interpolates `default_concurrency` against whatever config it was constructed with (`self._default_concurrency.eval(config=self.config)`)
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrency_level/concurrency_level.py#L39-L41
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrency_level/concurrency_level.py#L39-L41
- The resulting integer is passed straight into `ConcurrentSource.create(num_workers=concurrency_level, ...)`, which sizes the `ThreadPoolExecutor`
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L246-L247 and https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L61-L63
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L265-L266 and https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L61-L63
Audit of other raw `config` uses in `__init__`: the only other one is `get_registered_components_module(config=config)` (v7.23.8 https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L194, main https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L209). It runs before the spec component exists and reads only the injected `__injected_components_py` keys, which migrations do not touch, so it is not affected. `ConcurrencyLevel` is the only consumer of the stale config.
## Reproduction
Unit level, no network. Confirmed against a clean install of `airbyte-cdk==7.23.8`. Manifest fragment (wrap it in any otherwise valid minimal manifest: `version`, `check`, one stream):
```yaml
concurrency_level:
type: ConcurrencyLevel
default_concurrency: "{{ config.get('num_workers', 4) }}"
max_concurrency: 10
spec:
type: Spec
connection_specification:
type: object
properties:
num_workers:
type: integer
config_normalization_rules:
type: ConfigNormalizationRules
config_migrations:
- type: ConfigMigration
transformations:
- type: ConfigAddFields
condition: "{{ config.get('num_workers', 4) < 2 }}"
fields:
- type: AddedFieldDefinition
path: ["num_workers"]
value: "2"
value_type: integer
```
Build the source with `config={"num_workers": 1}` and inspect:
```python
source = ConcurrentDeclarativeSource(source_config=manifest, config={"num_workers": 1})
assert source._config["num_workers"] == 2 # migration applied
assert source._concurrent_source._threadpool._threadpool._max_workers == 1 # pool sized from the old value
```
Expected: both are 2. Actual: `self._config` is `{'num_workers': 2}` and `_max_workers` is 1. The `CONNECTOR_CONFIG` control message printed during construction carries `num_workers: 2`, so the discrepancy is only visible on the migrating run.
## Impact
Any connector that uses a config migration (or a per-sync config transformation) to raise, lower, or rename a field driving `default_concurrency` gets the wrong worker count for exactly one sync per connection: the first sync after the upgrade. For a migration that raises a floor (for example, to avoid a deadlock or starvation at 1 worker), that first sync still runs with the unsafe value. Symptoms are hard to diagnose because the emitted config looks correct and the second sync behaves as expected.
Concrete case: airbytehq/airbyte PR 85760 (open, source-zendesk-support 5.6.0) raises the minimum `num_workers` from 1 to 2 via `ConfigAddFields`. The migration alone did not protect the first post-upgrade sync, so the manifest also has to clamp inline:
```yaml
default_concurrency: "{{ [config.get('num_workers', 4), 2] | max }}"
```
That duplicates the migration logic in the manifest and is the workaround until the CDK is fixed.
## Suggested fix
In `ConcurrentDeclarativeSource.__init__`, pass the migrated config when building the component:
```python
concurrency_level_component = self._constructor.create_component(
model_type=ConcurrencyLevelModel,
component_definition=concurrency_level_from_manifest,
config=self._config,
)
```
`self._config` is already `{}` when no config was given, so `or {}` is not needed. Add a unit test in `unit_tests/sources/declarative/test_concurrent_declarative_source.py` (present at both refs: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/unit_tests/sources/declarative/test_concurrent_declarative_source.py, https://github.com/airbytehq/airbyte-python-cdk/blob/main/unit_tests/sources/declarative/test_concurrent_declarative_source.py) built from the manifest above, asserting that a field written by a `ConfigMigration` is visible to `concurrency_level.default_concurrency` on the same construction.
Once released, the inline clamp in source-zendesk-support can be reduced back to the plain `config.get('num_workers', 4)` expression.
## Precedent
- https://github.com/airbytehq/airbyte/pull/85760 - review of this PR surfaced the behaviour; the manifest-level clamp there is the current workaround.
- Internal incident reference: airbytehq/oncall#13250 (private repository).
贡献指南
调研方向
Start in airbyte_cdk/sources/declarative/concurrent_declarative_source.py, tracing __init__ through _migrate_and_transform_config and ConcurrencyLevel construction. Use the manifest reproduction in the issue, then add coverage in unit_tests/sources/declarative/test_concurrent_declarative_source.py. Done means a ConfigMigration-written value controls default_concurrency during the same source construction and the test passes.
由索引模型根据 Issue 内容生成。
评估
- 技术栈
- python
- 领域
- backend, testing
- Issue 类型
- 缺陷
- 难度
- 2/5
- 预计耗时
- 1-3 小时
- 活跃度
- 活跃
- 描述清晰度
- 描述清楚
- 新手友好度
- 88/100