Skip to content

Commit 020bdcc

Browse files
committed
Describe the all-handlers probe in ADR-0011 and ADR-0012
Both ADRs still had the request name the file's Dags and the runtime omit the rest. SDK authors and later layers read them as the spec for what a runtime answers, so they must say the answer depends only on the artifact.
1 parent 68e0cdd commit 020bdcc

2 files changed

Lines changed: 16 additions & 16 deletions

File tree

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -90,7 +90,7 @@ DagFileProcessorProcess(etl.py) ← manager spawn
9090
├── _serialize_dags(bag) → is_stub tasks carry arg_bindings (ADR-0007)
9191
│
9292
│ ┌─────────────────────────────────────────────────────────────────────┐
93-
│ │ Step 1: Collect the dag_ids to ask about — every Dag in this │
93+
│ │ Step 1: Collect the Dags to validate — every Dag in this │
9494
│ │ file with at least one is_stub task → ["etl"] │
9595
│ └─────────────────────────────────────────────────────────────────────┘
9696
│
@@ -123,7 +123,7 @@ DagFileProcessorProcess(etl.py) ← manager spawn
123123
│ │ └── BundleScanner.scanBundles(roots) │
124124
│ │ → "etl" → ResolvedBundle(analytics.jar, mainClass, ...) │
125125
│ │ │
126-
│ │ Group the dag_ids by (coordinator, artifact) — one group, one │
126+
│ │ Group the stub tasks by (coordinator, artifact) — one group, one │
127127
│ │ process, one request │
128128
│ └─────────────────────────────────────────────────────────────────────┘
129129
│
@@ -138,12 +138,11 @@ DagFileProcessorProcess(etl.py) ← manager spawn
138138
│ │ ├── in the child: _build_parse_task_handler_command() │
139139
│ │ │ coordinator.parse_task_handler() — spawn JVM │
140140
│ │ │ │
141-
│ │ │ ──TaskHandlerParseRequest(file=analytics.jar, │
142-
│ │ │ dag_ids=["etl"])─────▶ JVM │
141+
│ │ │ ──TaskHandlerParseRequest(file=analytics.jar)─────▶ JVM │
143142
│ │ │ (ToSDKTaskHandlerProcessor) │
144143
│ │ │ │
145-
│ │ │ JVM answers from its own TaskHandler registrations │
146-
│ │ │ whose dagId is one of the requested ids │
144+
│ │ │ JVM answers with every TaskHandler registration │
145+
│ │ │ in the artifact, or {} when there is none │
147146
│ │ │ │
148147
│ │ │ ◀─TaskHandlerParsingResult(task_handlers={ │
149148
│ │ │ "etl": [extract, transform, load]})─────── JVM │
@@ -176,8 +175,8 @@ DagFileProcessorProcess(etl.py) ← manager spawn
176175
```
177176

178177
The parse owns validation, not an importer. `PythonDagImporter` returns `airflow.sdk.DAG` objects and knows nothing about coordinators or queues, so `@task.stub` keeps working for
179-
any importer that can produce a Dag carrying stub tasks. `_parse_file` is also the only place where the whole file's Dags are visible at once, which is what lets one request cover
180-
every `dag_id` that resolved to the same artifact ([ADR-0012](0012-lang-sdk-parse-protocol.md)).
178+
any importer that can produce a Dag carrying stub tasks. `_parse_file` is also the only place where the whole file's Dags are visible at once, which is what lets one request per
179+
artifact serve every Dag in the file whose stubs resolved to it ([ADR-0012](0012-lang-sdk-parse-protocol.md)).
181180

182181
Resolution goes through the coordinator registry, not the filesystem, so the Python Dag and the Lang-SDK artifact **do not need to be in the same DagBundle**. Nothing here needs an
183182
`airflow.sdk.DAG` round-trip either — validation compares against the Dag the Python parser already built. Appendix B states exactly what is compared.
@@ -187,7 +186,7 @@ Resolution goes through the coordinator registry, not the filesystem, so the Pyt
187186
| Caller | Coordinator call | What comes back | Action |
188187
|--------------------------------------------|------------------------------------------------------|------------------------------------------|--------------------------------------------------|
189188
| `_parse_file` → `PythonDagImporter` | — (the Python file is parsed in process) | its own parsed Dags | PERSIST |
190-
| `_parse_file`, per (coordinator, artifact) | `parse_task_handler`, scoped to that group's dag_ids | `TaskHandlerParsingResult` | VALIDATE only — not a Dag, so nothing to persist |
189+
| `_parse_file`, per (coordinator, artifact) | `parse_task_handler`, for every handler it registers | `TaskHandlerParsingResult` | VALIDATE only — not a Dag, so nothing to persist |
191190
| `_parse_file` → `JavaDagImporter` | `parse_dag` | `DagFileParsingResult`, native Dags only | PERSIST |
192191

193192
There is no fourth row. A `TaskHandlerRef` has no Dag, so no `DagImporter` — and nothing reading a `DagImporter`'s results — ever sees one.
@@ -235,7 +234,8 @@ with one `bundle.serve()`. One artifact can hold both, so neither the file nor t
235234

236235
### Appendix B — What is compared
237236

238-
Handler declarations from every process spawned in Step 4 are unioned per `dag_id` before comparison, since one Dag's stubs can target several queues.
237+
Handler declarations from every process spawned in Step 4 are unioned per `dag_id` before comparison, since one Dag's stubs can target several queues. Each reply covers
238+
every Dag its artifact registers handlers for; only the parsed file's Dags are compared.
239239

240240
- `task_id` sets must match exactly. A missing or extra handler is an error.
241241
- `arg_bindings` against `params` as each declaration's `binding` says: by position for `positional`, by name for `named`, `named_or_whole` and `named_open`, compared

‎airflow-core/adr/lang-sdk/0012-lang-sdk-parse-protocol.md‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@ Proposed
2626
## Context
2727

2828
The Dag processor asks a Lang-SDK runtime two different questions. "Which Dags does this artifact define?" is answered over the messages [ADR-0004](0004-dag-parsing.md) already
29-
defines. "Which task handlers does this artifact register for a `dag_id` Python already owns?" has no answer in those messages, because a `TaskHandler` registration carries no Dag
30-
([ADR-0011](0011-mixed-language-dag-processing.md)).
29+
defines. "Which task handlers does this artifact register, each for a `dag_id` Python already owns?" has no answer in those messages, because a `TaskHandler` registration carries
30+
no Dag ([ADR-0011](0011-mixed-language-dag-processing.md)).
3131

3232
This ADR defines the request that carries the second question, the subprocess classes that carry both, and the two parse-side entry points on the coordinator.
3333

@@ -82,7 +82,6 @@ relay is only type-safe because both hops speak the same pair, which is the reas
8282
```
8383
class TaskHandlerParseRequest:
8484
file: str # the artifact resolved for this coordinator
85-
dag_ids: list[str] # every Dag in the parsed file with stub tasks that resolved here
8685
bundle_path: Path
8786
bundle_name: str
8887
type: Literal["TaskHandlerParseRequest"]
@@ -106,9 +105,10 @@ class TaskHandlerParam:
106105
exact_name: bool = False # match as spelled, not case-insensitively with underscores ignored
107106
```
108107

109-
One request carries every `dag_id` that resolved to the same artifact under the same coordinator, so a file whose stubs all target one runtime costs one process. A `dag_id` the
110-
artifact registers nothing for is **omitted** from `task_handlers` rather than returned empty: the key set is not required to match `dag_ids`, because it is the union across
111-
coordinators that has to cover the stubs ([ADR-0011](0011-mixed-language-dag-processing.md)).
108+
The request names no Dags. The runtime answers with every task handler the artifact registers, keyed by `dag_id`, and with `{}` when it registers none. The answer must depend only
109+
on the artifact, never on the request, so the Dag processor can cache it per artifact ([ADR-0013](0013-persisted-task-handler-bindings.md)). One request per (coordinator, artifact)
110+
pair serves every Dag in the parsed file whose stubs resolved there, so a file whose stubs all target one runtime costs one process. The key set is not required to match the file's
111+
Dags: it can hold other files' Dags, and it is the union across coordinators that has to cover the stubs ([ADR-0011](0011-mixed-language-dag-processing.md)).
112112

113113
`value_schema` reuses the `ArgValueSchema` definition `arg_bindings` already carries ([ADR-0007](0007-taskflow-across-language-boundary.md)), so both sides of a comparison are the
114114
same type. Two properties matter to validation: the field is nullable on both sides, and each declaration names its binding mode. Appendix B says what that forces.

0 commit comments

Comments
 (0)