jason810496 opened a new issue, #71134: URL: https://github.com/apache/airflow/issues/71134
### 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 -- 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]
