Add ADR-0008 for mixed-language Dag processing flow - #71929
Conversation
6324a01 to
ca8ebdd
Compare
| - [ADR-0005](0005-coordinator-packaging.md): coordinator packaging, module layout, and registration. | ||
| - [ADR-0006](0006-no-lang-sdk-source-display.md): no Lang-SDK source display for mixed-language (`@task.stub`) Dags. | ||
| - [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the language boundary — argument binding for Lang-SDK tasks. | ||
| - [ADR-0008](0008-mixed-language-dag-processing.md): mixed-language Dag processing — DagImporter routing and persistence. |
There was a problem hiding this comment.
#71189 also adds an ADR-0008 (0008-taskflow-dag-dsl.md) and appends to this same README line, so whichever of the two lands second needs renumbering to 0009 plus a README fixup. Since #71189 is still draft and this one isn't, simplest is probably to keep 0008 here and renumber there — but either way it's worth pinning down now.
| │ ┌─────────────────────────────────────────────────────────────────────┐ | ||
| │ │ Step 2: For each Coordinator, get its DagImporter and │ | ||
| │ │ list_dag_definitions to find the matching Dag │ | ||
| │ │ │ | ||
| │ │ JavaCoordinator(name="jdk-11") │ | ||
| │ │ │ │ | ||
| │ │ ├── Get artifact root: │ | ||
| │ │ │ coordinator.dag_bundle │ | ||
| │ │ │ → BaseDagBundle for "java-jdk11-bundle" │ | ||
| │ │ │ OR coordinator.artifact_roots │ | ||
| │ │ │ → [Path("/opt/airflow/jars/jdk11/")] │ | ||
| │ │ │ │ | ||
| │ │ ├── JavaDagImporter.list_dag_definitions( │ | ||
| │ │ │ bundle=coordinator.dag_bundle, │ | ||
| │ │ │ safe_mode=True) │ | ||
| │ │ │ → Iterator[DagDefinition] │ | ||
| │ │ │ DagDefinition("app-jdk11.jar") │ | ||
| │ │ │ DagDefinition("utils-jdk11.jar") │ | ||
| │ │ │ ... │ | ||
| │ │ │ │ | ||
| │ │ └── JavaDagImporter.import_definition( │ | ||
| │ │ definition, │ | ||
| │ │ bundle=coordinator.dag_bundle, │ | ||
| │ │ discover_mixed_language_dags=True) │ | ||
| │ │ → DagImportResult with Dag structure for dag_id="etl" │ | ||
| │ │ │ | ||
| │ │ JavaCoordinator(name="jdk-17") │ | ||
| │ │ └── (same flow, different artifact root / bundle) │ | ||
| │ └─────────────────────────────────────────────────────────────────────┘ |
There was a problem hiding this comment.
I don’t think it makes sense to use the importer here. The importer is designed to be used for dag processing, not on execution time. For stub-based definition, the only the Python importer should be needed; task implementation discovery (finding the JAR to use for a given task) should likely not use AIP-85. A standard may be built later, maybe based on the same interface as AIP-85, but I don’t think using AIP-85 here is a correct choice. Please leave this out; the existing discovery logic based on coordinator configuration should be used.
There was a problem hiding this comment.
To clarify, the entire flow described here occurs during Dag processing. It is not involved in task execution or in discovering an implementation when a task starts.
The proposed use of JavaDagImporter for mixed-language Dags has two goals:
- Reuse the AIP-85 DagImporter discovery interface instead of introducing a second interface with overlapping discovery responsibilities.
- Load the Lang-SDK-side Dag definition during processing so it can be compared with the Python stub-based definition. This allows mismatches to be reported as Dag import errors instead of surfacing later during task execution. We need
LazyDeserializedDAGhere for comparing the attributes, not just the dagId or taskId that the embedded airflow-metadata supports.
The Coordinator appears in the flow only because it provides the deployment-specific artifact location. The design needs to support both DagBundle-based deployment (#70805) and explicitly configured artifact roots. Resolving the Coordinator here does not invoke task execution or move this work to execution time.
jason810496
left a comment
There was a problem hiding this comment.
Here're the feedbacks suggested by TP from the offline sync:
- The Lang SDK user interface should be different for the mixed Lang Dag case (e.g.
ExternalDagorTaskBuilderinstead ofDagBuilderetc.) - Don't treat the mixed Lang Dag artifact same as the native Dag atifact.
- Instead of introducing the
is_mixed_languageDag-level argument, we need to adjust theDagFileParsingRequestto make the runtime subprocess propagate the mixed Lang Dag only list.
Design only, no code changes.
Was generative AI tooling used to co-author this PR?