meel-ayush opened a new issue, #76551:
URL: https://github.com/apache/airflow/issues/76551
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.4.0rc3 (and main)
### What happened and how to reproduce it?
### Description
In Airflow 3, `dag.dag_id` is the primary key on `DagModel` across all
configured DAG bundles (`bundle_name` is an attribute, not part of the primary
key). When two separate bundles (e.g. two git repositories) declare a DAG with
the same `dag_id`, parsing the second bundle unconditionally overwrites
`DagModel.bundle_name` and `DagModel.relative_fileloc`.
When the scheduler queues a task instance originally scheduled under the
first bundle, `ExecuteTask.make(ti)` builds a split-brain workload combining
the second bundle's name and file path with the first bundle's git commit SHA.
When the worker attempts to initialize the bundle repository, execution crashes:
```text
AirflowException: Version '1111111' not found in the repository for bundle
'bundle_beta'. Available versions: ['2222222']
```
In multi-team setups, this also reassigns task ownership to the second team,
causing the Execution API to reject the original team's worker with HTTP 403
Forbidden.
### Affected Code
In `airflow-core/src/airflow/dag_processing/collection.py:830-831`:
```python
dm.bundle_name = self.bundle_name
dm.bundle_version = self.bundle_version
```
And in `airflow-core/src/airflow/executors/workloads/task.py:122-124`:
```python
bundle_info = BundleInfo(
name=ti.dag_model.bundle_name, # Reads bundle_beta from
overwritten DagModel
version=ti.dag_run.bundle_version, # Reads bundle_alpha's git commit
SHA from DagRun
version_data=_resolve_version_data(ti.dag_run.created_dag_version,
ti.dag_run.bundle_version),
)
```
### Steps to Reproduce
1. Configure two versioned bundles: `bundle_alpha` (commit `1111111`) and
`bundle_beta` (commit `2222222`).
2. Add a DAG with `dag_id="shared_pipeline"` to both bundles.
3. Parse `bundle_alpha` and schedule a `DagRun`. A `TaskInstance` is queued
with `bundle_version="1111111"`.
4. Parse `bundle_beta`. `DagModel.bundle_name` is updated to `bundle_beta`.
5. The scheduler constructs the workload: `bundle_info.name = "bundle_beta"`
and `bundle_info.version = "1111111"`.
6. The worker tries to initialize `bundle_beta` with commit `1111111` and
crashes with `AirflowException`.
### Expected Behavior
A duplicate `dag_id` in a secondary bundle should not overwrite the
`bundle_name` or file location of an already registered DAG from another
bundle. The conflicting definition should be rejected with an import error or
ignored until resolved.
### Proposed Fix
In `airflow-core/src/airflow/dag_processing/collection.py`, check if
`existing.bundle_name != incoming_bundle` before updating `DagModel`. If a
duplicate DAG ID is detected from a different bundle, retain the existing
bundle association on `DagModel` and record a `ParseImportError` for the
secondary file.
I have verified this behavior with a local reproduction script and am
working on a PR with regression tests to fix this.
### What you think should happen instead?
_No response_
### Operating System
Kali GNU/Linux
### Deployment
None
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
