|
| 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: TaskFlow-Style Dag DSL for Native Java Dags |
| 21 | + |
| 22 | +## Status |
| 23 | + |
| 24 | +Accepted |
| 25 | + |
| 26 | +## Context |
| 27 | + |
| 28 | +The Go SDK adopted a TaskFlow-style authoring API: `Dag.Task(fn, opts...)` |
| 29 | +returns a handle, `Inputs(handle)` both wires the dependency edge and feeds the |
| 30 | +upstream's return value into the downstream function's data parameter, and |
| 31 | +`TaskSpec` / `DagSpec` configuration structs are generated from the Dag |
| 32 | +serialization schema (`airflow-core/src/airflow/serialization/schema.json`). |
| 33 | + |
| 34 | +The Java SDK could only implement *task bodies*. Everything else about the Dag |
| 35 | +had to come from a Python `@task.stub` file: |
| 36 | + |
| 37 | +- `@Builder.Dag` / `@Builder.Task` carried only identity attributes; there was |
| 38 | + no way to express schedule, retries, queue, or any other Dag/task |
| 39 | + configuration. |
| 40 | +- The Java-side model kept only an `id -> task class` map: no dependency edges |
| 41 | + and no configuration, so a Java author could not describe a graph and there |
| 42 | + was nothing to serialize for native Dag support. |
| 43 | + |
| 44 | +Python's TaskFlow shows the shape worth matching: calling tasks like functions |
| 45 | +*is* the graph declaration — `load(transform(extract()))`. Java annotations |
| 46 | +cannot change call semantics (invoking the real method would execute its body), |
| 47 | +so the call syntax has to target compile-time-generated twins. |
| 48 | + |
| 49 | +## Decision |
| 50 | + |
| 51 | +### Two authoring surfaces, one model |
| 52 | + |
| 53 | +**Annotation surface** — task bodies are `@Builder.Task`-annotated methods; |
| 54 | +configuration lives in the annotations; the graph is wired by a static |
| 55 | +`@Wiring` method that receives a processor-generated *flow twin* class |
| 56 | +(`<Class>Ref`). Twin methods mirror the task methods: injectable parameters |
| 57 | +(`Client`, `Context`) are dropped, data parameters become `In<T>`-typed |
| 58 | +inputs, and the return value becomes a `TaskRef<T>` (which extends |
| 59 | +`In<T>`). The call graph is the task graph: |
| 60 | + |
| 61 | +```java |
| 62 | +@Builder.Dag(id = "java_etl", schedule = "@daily") |
| 63 | +public class EtlPipeline { |
| 64 | + |
| 65 | + @Builder.Task(id = "extract", retries = 2) |
| 66 | + public ExtractResult extract(Client client) { ... } |
| 67 | + |
| 68 | + @Builder.Task |
| 69 | + public TransformResult transform(Client client, ExtractResult in) { ... } |
| 70 | + |
| 71 | + @Builder.Task |
| 72 | + public void load(Context ctx, TransformResult in) { ... } |
| 73 | + |
| 74 | + @Builder.Task |
| 75 | + public void score(double threshold) { ... } |
| 76 | + |
| 77 | + @Wiring |
| 78 | + static void depends(EtlPipelineRef f) { |
| 79 | + f.load(f.transform(f.extract())); |
| 80 | + f.score(In.value(0.5)); // inline literal input |
| 81 | + } |
| 82 | +} |
| 83 | +``` |
| 84 | + |
| 85 | +There is deliberately **no id-based wiring mode** and no edge-declaring API |
| 86 | +on the handles: the wiring calls, and only they, define the graph. A twin call |
| 87 | +registers the task, so a `@Builder.Task` method never invoked in the wiring |
| 88 | +method is an error (checked when the generated `build()` runs, at Dag-parse |
| 89 | +time). |
| 90 | + |
| 91 | +The wiring method is **optional**: a `@Builder.Dag` class without one registers |
| 92 | +every task with no Java-side edges. That is the shape for stub-backed tasks — |
| 93 | +the Python Dag file owns the graph and the supervisor delivers each call-site |
| 94 | +argument at run time — so those classes carry no `@Wiring`, no configuration |
| 95 | +attributes, and no Java-side dependencies at all. |
| 96 | + |
| 97 | +**Interface surface** — plain-class task definitions registered as first-class |
| 98 | +`TaskDef` objects, with fluent schema-validated `.config(key, value)` calls |
| 99 | +and object-reference edges: |
| 100 | + |
| 101 | +```java |
| 102 | +var extract = new TaskDef("extract", Extract.class).config("retries", 2); |
| 103 | +var transform = new TaskDef("transform", Transform.class).dependsOn(extract); |
| 104 | + |
| 105 | +var dag = new DagDef("java_etl") |
| 106 | + .config("schedule", "@daily") |
| 107 | + .addTask(extract) |
| 108 | + .addTask(transform); |
| 109 | +``` |
| 110 | + |
| 111 | +A `TaskDef` belongs to at most one Dag. Both surfaces flow into the same |
| 112 | +model: schema-keyed configuration maps, `TaskDef` dependency edges, and |
| 113 | +recorded parameter inputs on `DagDef`/`TaskDef`. |
| 114 | + |
| 115 | +### One package, one wildcard import |
| 116 | + |
| 117 | +The whole user-facing surface — the `Builder.Dag` / `Builder.Task` |
| 118 | +annotations, `@Wiring`, the flow types (`In`, `TaskRef`), and the runtime |
| 119 | +types (`DagDef`, `TaskDef`, `Task`, `Client`, `Context`, `Bundle`, ...) — |
| 120 | +lives in `org.apache.airflow.sdk`, so every example starts with a single |
| 121 | +`import org.apache.airflow.sdk.*`. There is no separate `dsl` package. |
| 122 | + |
| 123 | +Two top-level types cannot share a fully-qualified name, and `Dag` / `Task` |
| 124 | +are the names both the annotations and the runtime types want. Keeping the |
| 125 | +annotations **nested** in `Builder` resolves that without renaming anything: |
| 126 | +`Builder.Dag` and `Builder.Task` read as the annotations they are (they drive |
| 127 | +the `*Builder` codegen), the task-implementation interface keeps the plain |
| 128 | +name `Task`, and the Dag model is `DagDef`, symmetric with `TaskDef` — both |
| 129 | +are the definition objects of the interface API. `@Wiring` needs no |
| 130 | +qualification because nothing else claims that name. |
| 131 | + |
| 132 | +By convention the `@Wiring` method sits at the end of the Dag class, after |
| 133 | +the task methods it wires — read the tasks first, then the graph. |
| 134 | + |
| 135 | +### Wiring is type-checked by javac itself |
| 136 | + |
| 137 | +Twin input types make the graph checks ordinary Java type checking rather |
| 138 | +than bespoke processor analysis: |
| 139 | + |
| 140 | +- numeric parameters accept any numeric upstream (`In<? extends Number>`, |
| 141 | + widened or narrowed at run time by the shared decoder); |
| 142 | +- `Object`, raw `Map`, and raw `List` parameters accept any upstream |
| 143 | + (`In<?>`, decoded loosely at run time); |
| 144 | +- everything else accepts covariant matches of the declared type |
| 145 | + (`In<? extends T>`). |
| 146 | + |
| 147 | +Unknown upstreams are unrepresentable (a handle only exists once its task is |
| 148 | +registered), cycles are unconstructible in call syntax, and type mismatches |
| 149 | +are javac errors at the twin call site. The interface API's `dependsOn` can |
| 150 | +still express a cycle, so `Bundle` construction validates acyclicity (and |
| 151 | +that every referenced upstream is registered in the same Dag). |
| 152 | + |
| 153 | +### Runtime bindings win; wiring is the fallback |
| 154 | + |
| 155 | +Data parameters resolve by **position**, never by parameter name: Java call |
| 156 | +syntax is positional (no kwargs) and Java parameter names are not API — an |
| 157 | +IDE rename must not change binding behaviour. This matches the Go SDK's |
| 158 | +flat-parameter contract. |
| 159 | + |
| 160 | +When the supervisor delivered `arg_bindings` for the run, the binding at the |
| 161 | +parameter's position wins over anything the `@Wiring` method declared: for a |
| 162 | +stub task the Python call site *is* the graph the scheduler ordered the run |
| 163 | +by, so the Java class must not be able to disagree with it. The wiring- |
| 164 | +recorded inputs are the fallback, used when no bindings arrived — which is |
| 165 | +exactly the native-Dag case, where no Python call site exists. A `TaskRef` |
| 166 | +input then resolves to the upstream's return-value XCom and a literal input |
| 167 | +to its value. Keyword arguments still bind by name through a `TaskInput` |
| 168 | +bundle; the generated code branches on whether bindings arrived, because |
| 169 | +bindings fill the bundle field by field while the wiring fallback decodes the |
| 170 | +bundle wholesale from its single wired input. |
| 171 | + |
| 172 | +### Generated from the serialization schema |
| 173 | + |
| 174 | +The `Builder` class — the outer container plus its nested `Dag` and `Task` |
| 175 | +annotations, configuration attributes and all — and the `SchemaFields` |
| 176 | +validation table are generated at build time from a vendored copy of the Dag |
| 177 | +serialization schema (`sdk/schema/dag-schema.json`, kept in sync with |
| 178 | +airflow-core by the `sync-java-sdk-dag-schema` prek hook), mirroring the |
| 179 | +supervisor-schema → jsonschema2pojo pipeline that already exists. Generating |
| 180 | +the whole class rather than merging generated attributes into a hand-written |
| 181 | +one keeps a single definition of `Builder`; `id` (and `to` on `Dag`) stay the |
| 182 | +leading structural attributes, and generation fails if a schema key ever |
| 183 | +camel-cases onto one of them. |
| 184 | + |
| 185 | +Field selection mirrors the Go SDK's `TaskSpec` generator: scalar properties |
| 186 | +only, serializer-owned keys skipped (`_`-prefixed, schema-required, |
| 187 | +`has_on_*`), a documented exclusion list for Python-only concerns that fails |
| 188 | +generation when it goes stale, and a hand-curated Dag-level allowlist matching |
| 189 | +Go's `DagSpec`. `schedule` is a virtual key that the future serializer maps to |
| 190 | +the schema's `timetable` object. |
| 191 | + |
| 192 | +Temporal attributes are ISO-8601 strings in annotations (validated by the |
| 193 | +processor at compile time) and `java.time.Duration` / |
| 194 | +`java.time.OffsetDateTime` values in `.config` calls. |
| 195 | + |
| 196 | +### Explicit-only lowering, single-sourced dag id |
| 197 | + |
| 198 | +The processor lowers **only attributes written at the use site** into |
| 199 | +`DagDef.config(...)` / `TaskDef.config(...)` calls (via |
| 200 | +`AnnotationMirror.getElementValues`). Annotation defaults mirror schema |
| 201 | +defaults but are never emitted, so the serializer's omit-if-default semantics |
| 202 | +stay intact and the scheduler's own defaults win for everything unset. |
| 203 | + |
| 204 | +Because this written-vs-defaulted distinction only exists at compile time, |
| 205 | +there is no reflective `new DagDef(SomeClass.class)` constructor. Instead the |
| 206 | +generated builder exposes `DAG_ID` and a `dag()` factory (Dag-level config |
| 207 | +only, no tasks) alongside `build()`, so the dag id is never restated in user |
| 208 | +code. |
| 209 | + |
| 210 | +### No annotation on data parameters |
| 211 | + |
| 212 | +Data parameters need no annotation at all: anything that is not an injectable |
| 213 | +type (`Client`, `Context`) is a data parameter, bound in declaration order. |
| 214 | +Naming an upstream in an annotation would duplicate what the wiring — or the |
| 215 | +Python call site — already declares. `Client.getXCom` / `Client.setXCom` |
| 216 | +remain for imperative access. |
| 217 | + |
| 218 | +## Consequences |
| 219 | + |
| 220 | +- One concept, one name: bindings are *args*/*inputs* across the wire |
| 221 | + contract, the Go SDK, and the Java surface; the Java graph is declared the |
| 222 | + way Python TaskFlow declares it, but type-checked at compile time. |
| 223 | +- Dependency edges, configuration, and parameter inputs now exist in the |
| 224 | + Java-side model, which is the prerequisite for emitting |
| 225 | + DagSerialization-v3 JSON for native Java Dags. |
| 226 | +- Config typos fail at compile time (annotations) or Dag-parse time |
| 227 | + (`.config`), never silently. Wiring mistakes fail at compile time (type |
| 228 | + mismatch, unknown handle) or Dag-parse time (unregistered task, `dependsOn` |
| 229 | + cycle), never at task run time. |
| 230 | +- An annotation Dag class that owns its graph writes a `@Wiring` method — |
| 231 | + the same posture as the Go SDK, trading a few lines for one wiring story |
| 232 | + instead of two. Stub-backed classes omit it and stay registration-only. |
| 233 | +- The `@Wiring` method references the generated `<Class>Ref` twin, so IDEs |
| 234 | + show unresolved symbols until the first successful build (the standard |
| 235 | + Java codegen experience, as with Dagger or AutoValue). |
| 236 | +- New scalar schema keys show up automatically in the annotations and the |
| 237 | + validation table after a schema sync; removals fail the build until the |
| 238 | + exclusion rules are updated — the surface cannot drift silently. |
| 239 | +- Generated public API (annotation attributes) varies with the vendored |
| 240 | + schema version, exactly like the generated supervisor-schema models. |
0 commit comments