Skip to content

Commit 6324a01

Browse files
jason810496jason810496
committed
Add ADR-0008 for mixed-language Dag processing flow
1 parent 29c3ac2 commit 6324a01

2 files changed

Lines changed: 231 additions & 0 deletions

File tree

‎airflow-core/adr/lang-sdk/0008-mixed-language-dag-processing.md‎

Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,230 @@
1+
<!--
2+
Licensed to the Apache Software Foundation (ASF) under one
3+
or more contributor license agreements. See the NOTICE file
4+
distributed with this work for additional information
5+
regarding copyright ownership. The ASF licenses this file
6+
to you under the Apache License, Version 2.0 (the
7+
"License"); you may not use this file except in compliance
8+
with the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing,
13+
software distributed under the License is distributed on an
14+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
KIND, either express or implied. See the License for the
16+
specific language governing permissions and limitations
17+
under the License.
18+
-->
19+
20+
# ADR-0008: Mixed-Language Dag Processing — DagImporter Routing and Persistence
21+
22+
## Status
23+
24+
Proposed
25+
26+
## Context
27+
28+
A Lang-SDK artifact (JAR, packed Go executable) can serve two roles:
29+
30+
| Role | Dag defined in | Artifact contributes |
31+
|-------------------------|-------------------------------------|---------------------------|
32+
| **Mixed-language Dag** | Python file with `@task.stub` tasks | Only task implementations |
33+
| **Native Lang-SDK Dag** | The Go/Java source itself | The entire Dag |
34+
35+
Persisting both the Python Dag and the Lang-SDK Dag for the same `dag_id`
36+
creates conflicting definitions. The `is_mixed_language_dag` flag
37+
([PR #71213](https://github.com/apache/airflow/pull/71213)) disambiguates them.
38+
39+
This ADR uses `JavaDagImporter` as the concrete example. `ExecutableDagImporter`
40+
(Go) and `NodeDagImporter` (TypeScript) follow the same flow. A common base
41+
class is expected once the second Lang-SDK importer lands.
42+
43+
## Decision
44+
45+
### The Flag
46+
47+
```
48+
┌──────────────────────────────────────────────────────────────┐
49+
│ Per-task: is_stub │
50+
│ AbstractOperator.is_stub = False (default) │
51+
│ _StubOperator.is_stub = True │
52+
│ │
53+
│ Per-Dag: is_mixed_language_dag │
54+
│ Python side — derived at serialization: │
55+
│ if any(task.is_stub for task in dag.tasks): True │
56+
│ Lang-SDK side — set explicitly by the producer │
57+
└──────────────────────────────────────────────────────────────┘
58+
```
59+
60+
Both sides carry `is_mixed_language_dag: true`, so the flag alone cannot decide
61+
persistence. The decision is `(flag × importer)`.
62+
63+
### Processing Flows
64+
65+
#### Flow A — Pure Python Dag (no `@task.stub`)
66+
67+
```
68+
PythonDagImporter.import_definition(definition, bundle=...)
69+
│
70+
├── Parse → DAG objects
71+
├── serialize_dag(dag) → is_mixed_language_dag absent
72+
├── Return DagImportResult(dags=[dag])
73+
▼
74+
DagModelOperation → PERSIST
75+
```
76+
77+
#### Flow B — Lang-SDK Importer (native and mixed)
78+
79+
`JavaDagImporter` **never** persists a Dag with `is_mixed_language_dag: true`.
80+
81+
```
82+
JavaDagImporter.import_definition(definition, bundle=...)
83+
│
84+
├── BaseCoordinator.run_dag_parsing()
85+
│ └── Spawn JVM → DagFileParsingResult
86+
│ ┌─────────────────────────────────────────┐
87+
│ │ dag_id: "etl" │
88+
│ │ is_mixed_language_dag: true │
89+
│ ├─────────────────────────────────────────┤
90+
│ │ dag_id: "java_report" │
91+
│ │ is_mixed_language_dag: false │
92+
│ └─────────────────────────────────────────┘
93+
│
94+
├── "etl" is_mixed=true → DISCARD + log info
95+
├── "java_report" is_mixed=false → PERSIST
96+
│
97+
├── Return DagImportResult(dags=["java_report"])
98+
▼
99+
DagModelOperation → PERSIST "java_report" only
100+
```
101+
102+
#### Flow C — Mixed-Language Dag (validation driven by PythonDagImporter)
103+
104+
The `PythonDagImporter` owns validation. When it finds `is_mixed_language_dag:
105+
true`, it resolves each stub task's `queue` to its Coordinator, then uses the
106+
Coordinator's DagImporter to fetch the Lang-SDK Dag structure for
107+
cross-validation.
108+
109+
Resolution goes through the coordinator registry, not the filesystem — the Dag
110+
processor is file-at-a-time and the `PythonDagImporter` never sees the `.jar`.
111+
This also means the Python Dag and the Lang-SDK artifact **do not need to be in
112+
the same DagBundle**.
113+
114+
```
115+
PythonDagImporter.import_definition(definition, bundle=...)
116+
│
117+
├── Parse → DAG objects
118+
├── serialize_dag(dag) → is_mixed_language_dag = true
119+
│
120+
│ ┌─────────────────────────────────────────────────────────────────────┐
121+
│ │ Step 1: Resolve stub tasks → Coordinator instances via queue │
122+
│ │ │
123+
│ │ stub task queue Coordinator instance │
124+
│ │ ────────────────────────────────────────────────── │
125+
│ │ extract → "jdk-11" → JavaCoordinator(name="jdk-11") │
126+
│ │ transform → "jdk-17" → JavaCoordinator(name="jdk-17") │
127+
│ │ load → "jdk-11" → JavaCoordinator(name="jdk-11") │
128+
│ │ │
129+
│ │ Deduplicate → 2 distinct coordinator instances: │
130+
│ │ JavaCoordinator(name="jdk-11") │
131+
│ │ JavaCoordinator(name="jdk-17") │
132+
│ └─────────────────────────────────────────────────────────────────────┘
133+
│
134+
│ ┌─────────────────────────────────────────────────────────────────────┐
135+
│ │ Step 2: For each Coordinator, get its DagImporter and │
136+
│ │ list_dag_definitions to find the matching Dag │
137+
│ │ │
138+
│ │ JavaCoordinator(name="jdk-11") │
139+
│ │ │ │
140+
│ │ ├── Get artifact root: │
141+
│ │ │ coordinator.dag_bundle │
142+
│ │ │ → BaseDagBundle for "java-jdk11-bundle" │
143+
│ │ │ OR coordinator.artifact_roots │
144+
│ │ │ → [Path("/opt/airflow/jars/jdk11/")] │
145+
│ │ │ │
146+
│ │ ├── JavaDagImporter.list_dag_definitions( │
147+
│ │ │ bundle=coordinator.dag_bundle, │
148+
│ │ │ safe_mode=True) │
149+
│ │ │ → Iterator[DagDefinition] │
150+
│ │ │ DagDefinition("app-jdk11.jar") │
151+
│ │ │ DagDefinition("utils-jdk11.jar") │
152+
│ │ │ ... │
153+
│ │ │ │
154+
│ │ └── JavaDagImporter.import_definition( │
155+
│ │ definition, │
156+
│ │ bundle=coordinator.dag_bundle, │
157+
│ │ discover_mixed_language_dags=True) │
158+
│ │ → DagImportResult with Dag structure for dag_id="etl" │
159+
│ │ │
160+
│ │ JavaCoordinator(name="jdk-17") │
161+
│ │ └── (same flow, different artifact root / bundle) │
162+
│ └─────────────────────────────────────────────────────────────────────┘
163+
│
164+
│ ┌─────────────────────────────────────────────────────────────────────┐
165+
│ │ Step 3: Validate stub tasks against Lang-SDK tasks │
166+
│ │ │
167+
│ │ Python Dag "etl" Lang-SDK Dag "etl" │
168+
│ │ stub tasks: tasks: │
169+
│ │ extract ───────────── extract ✓ match │
170+
│ │ transform ─────────── transform ✓ match │
171+
│ │ load ──────────────── load ✓ match │
172+
│ │ │
173+
│ │ On mismatch → DagImportError │
174+
│ └─────────────────────────────────────────────────────────────────────┘
175+
│
176+
├── Validation passed → Return DagImportResult(dags=[dag])
177+
▼
178+
DagModelOperation → PERSIST (Python Dag is the sole DB record)
179+
```
180+
181+
### Decision Matrix
182+
183+
```
184+
┌───────────────────┬───────────────────────┬──────────────────────────────────────────────────────────────┐
185+
│ Importer │ is_mixed_language_dag │ Action │
186+
├───────────────────┼───────────────────────┼──────────────────────────────────────────────────────────────┤
187+
│ PythonDagImporter │ false │ PERSIST │
188+
├───────────────────┼───────────────────────┼──────────────────────────────────────────────────────────────┤
189+
│ PythonDagImporter │ true │ PERSIST + VALIDATE via queue → Coordinator → │
190+
│ │ │ DagImporter.list_dag_definitions + import_definition │
191+
├───────────────────┼───────────────────────┼──────────────────────────────────────────────────────────────┤
192+
│ JavaDagImporter │ false (or absent) │ PERSIST │
193+
├───────────────────┼───────────────────────┼──────────────────────────────────────────────────────────────┤
194+
│ JavaDagImporter │ true │ DISCARD + log info │
195+
└───────────────────┴───────────────────────┴──────────────────────────────────────────────────────────────┘
196+
```
197+
198+
### Why the Flag Lives on the Dag, Not the Bundle
199+
200+
A single artifact can contain both native and mixed-language Dags:
201+
202+
```
203+
analytics.jar
204+
├── dag_id="etl" is_mixed_language_dag=true (backs etl.py)
205+
└── dag_id="java_report" is_mixed_language_dag=false (native)
206+
```
207+
208+
The importer decides per-Dag, not per-file or per-bundle.
209+
210+
## Consequences
211+
212+
- **Python leads, Lang-SDK follows.** `PythonDagImporter` persists the Dag and
213+
drives validation via `queue → Coordinator → DagImporter.list_dag_definitions
214+
→ import_definition`. Lang-SDK importers silently discard mixed-language Dags.
215+
- Stub/implementation mismatches surface as `DagImportError` at parse time.
216+
- The Python Dag and Lang-SDK artifact can live in different DagBundles —
217+
resolution goes through the coordinator registry, not the filesystem.
218+
- A single Dag can have stubs targeting different queues (some Java, some Go) —
219+
each resolves to its own coordinator instance independently.
220+
- No schema migration, no new `DagModel` column, no REST/UI change.
221+
222+
## References
223+
224+
- [ADR-0004](0004-dag-parsing.md) — `BaseCoordinator` / `DagFileParsingResult`
225+
- [ADR-0006](0006-no-lang-sdk-source-display.md) — no Lang-SDK source display
226+
for mixed-language Dags
227+
- [AIP-108](https://cwiki.apache.org/confluence/x/pY4mGQ) — Language SDKs
228+
- [AIP-85](https://cwiki.apache.org/confluence/x/_Q7OEg) — DagImporter
229+
- [PR #71213](https://github.com/apache/airflow/pull/71213) — the
230+
`is_mixed_language_dag` flag

‎airflow-core/adr/lang-sdk/README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ bind core interfaces and apply to every language SDK, not just the Java SDK.
3333
- [ADR-0005](0005-coordinator-packaging.md): coordinator packaging, module layout, and registration.
3434
- [ADR-0006](0006-no-lang-sdk-source-display.md): no Lang-SDK source display for mixed-language (`@task.stub`) Dags.
3535
- [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the language boundary — argument binding for Lang-SDK tasks.
36+
- [ADR-0008](0008-mixed-language-dag-processing.md): mixed-language Dag processing — DagImporter routing and persistence.
3637

3738
Decisions specific to a single SDK stay next to that SDK — for example, the Go SDK's bundle-format
3839
decisions live in [`go-sdk/adr/`](../../../go-sdk/adr).

0 commit comments

Comments
 (0)