Coordinator: detect ambiguous entrypoints and duplicate dag_ids at import time
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 7h
- Merged PRs (30d)
- 484
Description
### Background
Subprocess coordinators can now locate language-SDK artifacts through a Dag bundle instead of
an explicit filesystem root (#70805). In the co-located default — no explicit root, no
`dag_bundle_name` — the coordinator scans the task's own Dag bundle.
For `JavaCoordinator` that makes an existing ambiguity the normal case rather than an edge case.
With `main_class` unset, `_JarInfo.find` returns the **first** JAR carrying a `Main-Class` in walk
order, and `_calculate_classpath` puts every JAR it found on the classpath:
https://github.com/apache/airflow/blob/3a08a3d7792335ac9456cd44c92d8de3ae0d6ec1/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py#L126-L144
So a deployment that ships more than one executable JAR in a bundle gets a non-deterministic
entrypoint, and if two of those JARs declare the same `dag_id`, which definition actually runs
is decided by directory walk order. Nothing warns the user; the task simply runs whichever one
was reached first.
The Python path already handles the equivalent case: `_build_duplicate_dag_id_warnings` in
`airflow-core/src/airflow/dag_processing/collection.py:327` detects a `dag_id` registered from
more than one file and emits a `DagWarning(DUPLICATE_DAG_ID)`. Language-SDK Dags have no
equivalent, because parsing happens inside the subprocess and there is no import-stage channel
to report into.
#70805 documents the gap rather than closing it — it adds a TODO in
`_build_execute_task_command` (`task-sdk/src/airflow/sdk/coordinators/java/coordinator.py`) and a
note in `airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst` telling users to set
`main_class` explicitly:
```python
# TODO: Scanning a whole Dag bundle without an explicit main_class lets
# _JarInfo.find pick the first executable JAR in walk order, so duplicate
# entrypoints across bundles resolve non-deterministically — the same
# duplicate-dag_id ambiguity the Python Dag path has. Reject it at the
# IMPORT_ERROR stage once AIP-85 exposes an interface to raise there.
jar = _JarInfo.find(roots, self.main_class)
```
### Why this is on hold
The right place to report this is at parse/import time, not at task launch — by the time
`_build_execute_task_command` runs, the Dag is already scheduled and the user gets, at best, a
task failure with no explanation of the ambiguity.
AIP-85's importer layer is where that reporting belongs.
`airflow-core/src/airflow/dag_processing/importers/base.py` already models `DagImportError`,
`DagImportWarning`, and `DagImportResult`, but `DagImporterRegistry` currently only registers
`PythonDagImporter`, and there is no path for a coordinator-backed parse to raise into the
import-error stage. There is nothing to build against until AIP-85 lands the non-Python importer
path, so this is deliberately parked rather than worked around further.
### What needs to happen
1. Wait for AIP-85 to land a language-SDK importer path and an interface that lets a
coordinator-backed parse report `DagImportError` / `DagImportWarning`.
2. During the bundle scan, detect **ambiguous entrypoints**: more than one artifact declaring an
entrypoint (`Main-Class` for Java) while `main_class` is unset.
3. During the bundle scan, detect **duplicate `dag_id`s**: the same `dag_id` declared by more
than one artifact under the resolved scan root.
4. Decide the severity for each and keep it consistent with the Python path — duplicate `dag_id`
is a `DagWarning(DUPLICATE_DAG_ID)` there, whereas an unresolvable entrypoint is arguably a
hard `DagImportError`.
5. Apply the same detection to the other subprocess coordinators (`ExecutableCoordinator`,
`NodeCoordinator`). The walk-order ambiguity is not Java-specific once co-located mode scans a
whole bundle; Java is only the loudest case because it also affects the classpath.
6. Remove the TODO in `_build_execute_task_command` and tighten the `java.rst` note once the
check exists.
### Acceptance criteria
- Deploying two JARs that declare the same `dag_id` into one bundle surfaces a visible
import-time error/warning in the UI instead of silently resolving to one of them.
- With `main_class` unset and multiple executable JARs in the resolved scan root, the ambiguity is
reported at parse time rather than resolved arbitrarily at task launch.
- The reported severity and message are consistent with the Python duplicate-`dag_id` behaviour.
- The TODO in `task-sdk/src/airflow/sdk/coordinators/java/coordinator.py` is removed and the
`java.rst` warning updated to describe the enforced behaviour.
- Tests cover both the duplicate-entrypoint and duplicate-`dag_id` cases.
### Context
- Review thread that raised this: https://github.com/apache/airflow/pull/70805#discussion_r3690030853
- PR that introduced the co-located bundle scan: #70805
- Related: #66334 (Java-SDK artifact deployment strategies)
- AIP-85: https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/315494137/AIP-85+DAG+importer
Contributor guide
Assessment
This issue has not been assessed yet.