jason810496 opened a new pull request, #71189: URL: https://github.com/apache/airflow/pull/71189
related: AIP-108 depends on: #69757, #71057 ## Why Until now the Java SDK could only supply task *bodies*. A Python `@task.stub` Dag had to own the schedule, every task option, and the graph — so a pipeline whose logic is entirely Java still had to be described in two languages, in two places, for no reason other than a missing authoring surface. It also left the Java-side model with nothing to describe: `DagDef` held an `id -> task class` map, no edges and no configuration, so there was nothing a native Java Dag could be built from and nothing to serialize for one. Python's TaskFlow shows the shape worth matching: calling tasks like functions *is* the graph declaration — `load(transform(extract()))`. Java annotations cannot change call semantics the way Python decorators do (invoking the real method would run its body), so the call syntax has to target something generated at compile time. ## How - **The graph is declared by calling processor-generated twins.** A static `@Wiring` method receives a generated `<Class>Ref` class whose methods mirror the `@Builder.Task` methods: injected `Client`/`Context` parameters are dropped, data parameters take `In<T>`, and the return value is a `TaskRef<T>`. Calling a twin registers the task; passing one twin's handle into another feeds the upstream's output into the downstream's parameter *and* records the edge. The call graph is the task graph, and `javac` checks it — numeric parameters accept any numeric upstream (`In<? extends Number>`), `Object`/raw `Map`/raw `List` accept any (`In<?>`), everything else accepts covariant matches. Unknown upstreams are unrepresentable and cycles are unconstructible in call syntax. - **The wiring calls are the only way to declare an edge** in the annotation API: no id-based mode, no edge API on the handles. One graph story, not two. A `@Builder.Task` method the wiring never invoked fails at Dag-parse time. - **`@Wiring` is optional, so stub-backed classes are untouched.** A class without one registers every task with no Java-side edges — exactly today's behaviour. The existing stub-backed examples gain neither `@Wiring` nor configuration. - **Runtime bindings keep winning over Java-declared wiring.** When the supervisor delivered `arg_bindings`, the binding at a parameter's position is what the task receives; for a stub task the Python call site is the graph the scheduler ordered the run by, so the Java class must not be able to disagree with it. Wired inputs are the fallback — the native-Dag case, where no Python call site exists. Binding is positional either way: Java parameter names are not API, so an IDE rename must not rebind an input. - **Configuration is generated from Airflow's Dag serialization schema**, not hand-listed, so the Java attributes cannot drift from the Python semantics they mirror and new scalar keys appear after a schema sync. Field selection mirrors the Go SDK's `TaskSpec` generator (scalars only, serializer-owned keys skipped, a documented exclusion list that fails generation when it goes stale, a hand-curated Dag-level allowlist). Only attributes *written at the use site* are lowered into `config` calls, so Airflow's own defaults still govern everything left out. - **The whole `Builder` class is generated**, outer class and both nested annotations, so there is exactly one definition of it; `id` (and `to` on `Dag`) stay the leading structural attributes, and generation fails if a schema key ever camel-cases onto one of them. - **`Bundle` construction validates what the type system cannot.** `TaskDef.dependsOn` can express a cycle or point at a task in another Dag, so the bundle checks acyclicity and same-Dag upstream membership at parse time. ## What - Add `@Wiring`, `In`/`TaskRef`, `internal.Refs` (twin registration), and `internal.Fields` (config validation). - Generate `Builder` (with schema-derived `@Builder.Dag` / `@Builder.Task` configuration attributes) and `internal.SchemaFields` from a vendored `sdk/schema/dag-schema.json`, kept in sync with `airflow-core` by the new `sync-java-sdk-dag-schema` prek hook; delete the hand-written `Builder.kt`. - `BuilderProcessor` emits the `<Class>Ref` twin, a `DAG_ID` constant and a `dag()` factory, lowers explicit annotation attributes into `config` calls (validating ISO-8601 temporals at compile time), and verifies the wiring registered every task. - `DagDef.config`, `TaskDef.config` / `dependsOn` / `inputs`, an `addTask(task, upstreams)` overload, cycle and upstream validation in `Bundle`, and `Context.taskDef` threaded through the task runner. - `ArgValues` falls back to the `@Wiring`-recorded inputs when the supervisor sent no bindings, and exposes `hasRuntimeBindings` so a `TaskInput` bundle is filled field-by-field from bindings but decoded wholesale from its single wired input otherwise. - New `nativedag/` examples in both styles (registered in the example bundle), the ADR at `airflow-core/adr/lang-sdk/0007-taskflow-dag-dsl.md`, and a "Native Java Dags" section in the Java SDK docs. --- ##### Was generative AI tooling used to co-author this PR? - [x] Yes, with help of Claude Code Opus 5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
