java-sdk/adr/0002-native-dag-interface.md

<!--
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

   http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
 -->

# ADR-0002: Native Java Dag — Interface Design

## Status

Proposed

## Context

A Dag authored with no Python stub file has no `@task.stub` call site to declare its graph, so
Java itself must express the graph, the Dag/task configuration, and the task bodies. This ADR is
scoped to what that Java call site looks like for a user. It also settles the injectable
`client`/`context` question shared with [ADR-0001](0001-mixed-lang-dag-interface.md): they are
**injected as method arguments**, not exposed through getters. It shares the protocol substrate
(the argument-binding spec) with
[`airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md`](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md).

## Decision

### Annotation based

`Client` and `Context` are injected as **method arguments**, exactly as on the mixed-language
surface ([ADR-0001](0001-mixed-lang-dag-interface.md)) — there are no getters and no SDK base class
to extend. A task method's signature is its injected arguments, if any, followed by its data:

```java
@Builder.Dag(id = "java_etl", schedule = "@daily")
public class EtlPipeline { // extends nothing of ours; your own base class stays free

  @Builder.Task(id = "extract", retries = 2)
  public long extract(Client client) {
    return ((Number) client.getVariable("row_count")).longValue();
  }

  @Builder.Task(id = "transform")
  public long transform(Client client, Context context, long extracted, double threshold) {
    client.setXCom("scaled_in", context.runId);
    return (long) (extracted * threshold);
  }

  @Builder.Task(id = "load")
  public void load(Context context, long transformed) {
    log.log(INFO, "Loaded {0} for run {1}", transformed, context.runId);
  }

  @Builder.Task(id = "audit")
  public void audit(Client client) { /* side effect only, no data in or out */ }

  @Builder.Task(id = "notify")
  // "notify" is Object.notify, so the method takes another name and keeps the task id.
  public void alert(Client client) { /* side effect only, no data in or out */ }

  @Builder.Deps
  static class Wiring implements EtlPipelineDeps {
    void depends() {
      // TaskFlow (data) edges: implied by passing a TaskRef; a constant is wrapped with lit(...)
      var rows = extract();
      var scaled = transform(rows, lit(0.9));
      var loaded = load(scaled);

      // non-TaskFlow (ordering-only) edges: sequence with no data flowing
      rows.then(audit());              // extract >> audit
      Flow.of(loaded, audit()).then(alert()); // [load, audit] >> notify
    }
  }

  public static void main(String[] args) {
    Bundle bundle = new Bundle().register(EtlPipeline.class);
    Server.create(args).serve(bundle);
  }
}
```

Because the injected arguments are real parameters, the wiring expression cannot call the task
methods directly — `extract(client)` would demand a `Client` the wiring has no business supplying.
So the processor generates one **wiring view** per Dag: the interface `<Dag>Deps` (here
`EtlPipelineDeps`), one default method per task with the injected arguments stripped, each data
argument lifted to `Arg<T>`, and the return lifted to `TaskRef<T>`:

```java
// Generated by the Airflow Java SDK annotation processor. Do not edit.
interface EtlPipelineDeps extends Deps { // Deps: the shared base that nests Flow
  default TaskRef<Long> extract() { return Flow.node("extract"); }
  default TaskRef<Long> transform(Arg<Long> extracted, Arg<Double> threshold) { return Flow.call("transform", extracted, threshold); }
  default TaskRef<Void> load(Arg<Long> transformed) { return Flow.call("load", transformed); }
  default TaskRef<Void> audit() { return Flow.node("audit"); }
  default TaskRef<Void> alert() { return Flow.node("notify"); }
}
```

The `@Builder.Deps` class implements this interface, so `depends()` composes the wiring methods and
`javac` type-checks the graph — `extract()` yields a `TaskRef<Long>`, which is an `Arg<Long>` for
`transform`, and so on. The class only *implements an interface*, so its `extends` stays free, and
being `static` it cannot reach the outer class's real task methods — wiring can only touch the
generated views.

`@Builder.Deps` is read by executing it once, in **recording mode**, at registration: each wiring
call records a memoized node (so `extract()` is the same `TaskRef` everywhere), passing a `TaskRef`
into another call records a data edge, and the ordering verbs below record ordering edges. No task
body runs, and — because a `TaskRef` carries the node's identity as an object reference — the
processor never reads the method's syntax tree, so this works under any compiler, not only javac.

The result is the Dag below, equivalent to the interface-based form written by hand:

```java
// Generated by the Airflow Java SDK annotation processor. Do not edit.
public final class EtlPipeline_Dag {

  public static Dag dag() {
    Dag dag = new Dag("java_etl").config("schedule", "@daily");
    TaskRef extract = dag.task("extract", Extract.class).config("retries", 2);
    TaskRef transform = dag.task("transform", Transform.class);
    TaskRef load = dag.task("load", Load.class);
    TaskRef audit = dag.task("audit", Audit.class);
    TaskRef notify = dag.task("notify", Notify.class);
    extract.then(transform).then(load); // data path
    extract.then(audit);                // ordering-only edge
    Flow.of(load, audit).then(notify);  // ordering-only edge
    return dag;
  }

  public static final class Transform implements Task {
    @Override
    public void execute(Context context, Client client) throws Exception {
      TaskArgs args = TaskArgs.of(context, client, 1);
      long extracted = args.require(0, Long.class);
      double threshold = 0.9; // baked from lit(0.9) at Dag-build time
      client.setXCom(new EtlPipeline().transform(client, context, extracted, threshold));
    }
  }

  // Extract, Load, Audit, and Notify follow the same shape.
}
```

The injected arguments are passed straight into the user method, and the data arguments bind by
position through the same internal `TaskArgs` the mixed-language surface uses
([ADR-0001](0001-mixed-lang-dag-interface.md)) — no getters, no `bind()`, nothing ambient.

**Literals.** A data argument is an `Arg<T>`, which a `TaskRef` satisfies; a constant is wrapped
with `lit(...)` — `transform(extract(), lit(0.9))` — and recorded as a baked value with no edge.
Wrapping is required because a bare `Integer` cannot implement `Arg`; boxed types only, no
primitives.

### Non-data dependencies

A data edge comes for free from passing a `TaskRef` into another wiring method. A dependency
where no data flows, similar to Python's `a >> b`, is expressed with `then`, a variadic verb
that every `TaskRef` carries. It lives on a small `Flow` interface, and a `TaskRef` is a `Flow`
of one (`interface TaskRef<T> extends Arg<T>, Flow`):

```java
a.then(b);     // a >> b
a.then(b, c);  // a >> [b, c]
```

`then` returns the **new frontier** (the set it just pointed at), the way `>>` evaluates to its
right operand, so a chain walks through a fan:

```java
a.then(b, c).then(d); // a >> [b, c] >> d  (a->b, a->c, then b->d, c->d)
```

The one thing `then` cannot do is start from a *set*: Java can't overload `>>` the way Python
does, and there is no list literal to call `.then` on, so `Flow.of` opens a chain from one:

```java
Flow.of(a, b).then(c);    // [a, b] >> c
Flow.of(a, b).then(c, d); // [a, b] >> [c, d]
```

Both edge kinds compile to the same `then(...)` in the generated Dag above. The only difference is
that a data edge also binds an argument while an ordering-only edge binds nothing.

### Interface based

```java
Dag dag = new Dag("java_etl").config("schedule", "@daily");

TaskRef extract = dag.task("extract", Extract.class).config("retries", 2);
TaskRef transform = dag.task("transform", Transform.class);
TaskRef load = dag.task("load", Load.class);
TaskRef notify = dag.task("notify", Notify.class);

extract.then(transform).then(load).then(notify);
```

`dag.task(id, class)` registers the task as it creates it and hands back a `TaskRef`, so there is no
second `addTask(...)` call to forget. `then` is the same variadic edge verb as the
annotation surface — Java's spelling of Python's `>>` — and `.config(key, value)` carries
Dag and task configuration. This surface wires edges through object references and reads no syntax
tree, so it works under any toolchain and is exactly what the annotation surface's recording
produces.

### Registering and serving

Both surfaces produce a `Dag`, and a `Bundle` is what the server serves — create it, register, serve:

```java
public static void main(String[] args) {
  Bundle bundle = new Bundle()
      .register(EtlPipeline.class) // annotation based: the user's own annotated class
      .register(dag);              // interface based: the Dag built above

  Server.create(args).serve(bundle);
}
```

An annotated Dag registers as the class the user wrote — the same class the annotations are on — so
there is no second name to learn or keep in sync.

Nothing is constructed at registration. Task classes are instantiated per task-instance
run ([ADR-0001](0001-mixed-lang-dag-interface.md)), and the Dag definition itself is read once, at
registration, without running a task body.

One bundle carries native Dags and mixed-language task handlers alike
([ADR-0001](0001-mixed-lang-dag-interface.md)), so one process serves both.

## Alternatives

- **Getters for `Client`/`Context`** (`getClient()`/`getContext()` on an SDK base class the Dag
  extends), so a task method takes only its data and `depends()` composes the real signatures
  directly, with no wiring view. Rejected: the team does not want ambient getters — they invite the
  "reach for the context anywhere" pattern we already avoid in Python — and a base class spends the
  Dag class's single `extends` on the SDK, so the user can no longer extend their own. Injected
  arguments keep the getter out of the language and the Dag class's inheritance free; the wiring view
  is the price, and it is generated rather than written.

- **Reading the `@Wiring` body's syntax tree** instead of executing it. With getters the task
  methods are data-only, so `depends()` composes real signatures and a processor could walk the tree
  and emit one edge per nested call. Rejected on two counts: it depends on
  `com.sun.source.util.Trees`, javac's tree model, so it does not run under a compiler with its own
  (ecj); and it must recover identity syntactically — a result held in a local
  (`long e = extract(); fanin(e, ...)`) is an identifier, not a call, so the walker has to resolve it
  through the initializer and bail on anything it cannot attribute. Recording sidesteps both: nothing
  parses source, and a reused local just works.

- **Capturing the graph by overriding task methods in a generated subclass**, so an inherited
  `depends()` calls the overrides. Rejected because a raw return (`long`) carries no identity, so a
  result stored in a local and used twice cannot be attributed to its producer. Recording avoids this
  precisely by having the wiring view return `TaskRef` rather than the task's real return type —
  identity travels with the handle, so this is not the same mechanism as the subclass capture.

- **Naming the type `Flow`, over `Order` and `Chain`.** The type every `TaskRef` carries, and that
  `Flow.of` returns, needed a name. `Chain` was rejected because Airflow's Python `chain()` is a
  *sequential* helper (`chain(a, b)` means `a >> b`), so `Chain.of(a, b)` would read as the edge
  `a -> b` rather than the parallel set `{a, b}`. `Order` reads well for pure sequencing, but misnames
  a group that can carry data: a chain like `Flow.of(a(), b(c())).then(d(e()))` also records the data
  edges `c -> b` and `e -> d`, so the group is tasks *in the flow*, not an ordering. It shares a
  name with `java.util.concurrent.Flow`, but nesting defuses that: `Flow` is a member type of the
  shared `Deps` base every `<Dag>Deps` extends, so wiring code inherits it by simple name and never
  imports it (see Implementation Notes).

- **Only `then`, not Python's fuller set of relation helpers.** Python offers `>>` and `<<`,
  `chain()`, etc. This surface has only one forward verb, plus `Flow.of`. A single left-to-right verb reads
  consistently, and is less to learn. Nothing is lost; the reverse direction and the shorthands can still be
  properly expressed.

## Consequences

- Two authoring surfaces (annotation, interface) flow into the same `Dag`/`TaskRef` model, and the
  annotation surface's recording produces exactly the interface-based calls a user could have
  written.
- `Client`/`Context` are injected as arguments on both the native and mixed-language surfaces, so a
  single injection rule spans the SDK and no getter or SDK base class exists.
- The wiring view (`<Dag>Deps`) is generated, so the graph is real, type-checked Java — but the
  interface name resolves in an IDE only after the first build that runs the processor. This is the
  standard annotation-processing tradeoff (Dagger, AutoValue, MapStruct) and needs the processor
  wired into the build so generated sources are indexed.
- Recording reads no syntax tree, so the annotation surface works under any compiler, not only javac.
- A task method keeps its real signature — injected arguments then data — and the only generated
  user-facing type is the wiring-view interface it implements: no twin class, no `In<T>`, no getters,
  no `bind()`.
- Native tasks keep `@Builder.Task`, while the mixed-language surface uses
  `@Builder.TaskHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names which
  of the two a method is.

## Appendix: Implementation Notes

- The processor generates, per Dag, the `<Dag>Deps` wiring-view interface and a registrar. The
  registrar instantiates the `@Builder.Deps` class, installs a recorder, calls `depends()`, collects
  the graph, and validates it (no cycles, no self-edges), reporting a bad edge on its own source line
  through `Messager`. It reads no syntax tree.
- The wiring view strips a parameter only when its type is `Client` or `Context`; every other
  parameter becomes an `Arg<T>` in declared order, and the return becomes `TaskRef<T>`
  (`TaskRef<Void>` for `void`).
- **The nested `@Builder.Deps` class is required, not stylistic.** The Dag class holds
  `long extract(Client)` and the wiring view holds `TaskRef<Long> extract()` — two methods that would
  be an overload clash in one class. A separate type carries the view, and it `implements` (never
  `extends`) so the Dag class keeps its single inheritance slot.
- Wiring methods are memoized by task id, so `extract()` returns the same `TaskRef` wherever it is
  called; that identity is what lets a data edge and a later ordering edge refer to one node.
- **`then` is variadic and returns the new frontier**, so `a.then(b, c).then(d)` walks through a
  fan, and linear runs, fan-outs, and fan-ins all fall out of the one verb. `Flow.of(...)` is the
  only entry point beyond it — it opens a chain from a set, which `then` cannot. A data edge never
  needs any of this — it is implied by passing a `TaskRef`.
- **`Flow` is a nested member of a shared base, not a top-level type.** Every generated `<Dag>Deps`
  extends one library interface, `Deps`, that declares the nested interface `Flow`. Member types are
  inherited, so a `@Builder.Deps` class implementing `<Dag>Deps` names `Flow` (as in `Flow.of(...)`)
  by simple name with no import — so there is nothing for `java.util.concurrent.Flow` to clash with.
  A single-type-import does not override this: a visible member type outranks a single-type-import in
  simple-name resolution (JLS §6.5.5.1; and §6.4.1, whose single-type-import shadowing list covers
  only top-level types in other units and on-demand imports, not member types), so even an explicit
  `import java.util.concurrent.Flow` in the same file still resolves `Flow` to the inherited member.
  The imperative interface-based surface does not extend `Deps`, so it qualifies the factory as
  `Deps.Flow.of(...)` — or, since it always has named refs, skips it entirely, writing a fan-in as
  `x.then(z); y.then(z)`.
- **`Arg<T>` is the data-argument type in a wiring method**; `TaskRef<T> extends Arg<T>`, and
  `lit(value)` wraps a boxed constant. A `TaskRef` argument records an edge; a `lit(...)` records a
  baked value with none. Because both flow through `Arg` and the recorder distinguishes them by type,
  there is no ambiguity even at an `Object`-typed position.
- **At run time the getters are gone.** The generated `Task` passes the injected `Client`/`Context`
  straight into the user method and binds data arguments by position through `TaskArgs`
  ([ADR-0001](0001-mixed-lang-dag-interface.md)) — the same conversion-not-cast path, including the
  `TypeReference<T>` overload for generics. Nothing is ambient, so the SDK's Java 11 pin (no
  `ScopedValue`) does not matter here.
- **`dag.task(...)`, `.config(...)`, and `.then(...)` are proposed additions.**
  Today's shipped `Dag` has only `addTask(id, definition)`, which returns the Dag rather than a ref;
  `.config(...)` is keyed to the Dag serialization schema. `dag.task(...)` is a factory method, not a
  constructor — Java spells qualified inner-class creation `dag.new TaskRef(...)`, which reads as
  compiler trivia rather than as a Dag definition.
- **`register` is one overloaded verb**: `register(Class<?>)` for an annotated Dag class,
  `register(Dag)` for a built one, and `register(dag, task, Class<? extends Task>)` for a task
  handler ([ADR-0001](0001-mixed-lang-dag-interface.md)). Today a bundle is built from an
  `Iterable<Dag>` through `BundleBuilder.getDags()`.
- **Registration holds classes, never instances.** `Dag.addTask(id, Class<? extends Task>)` already
  states that Airflow "instantiates the class via its no-argument constructor, then calls `execute`
  once per task-instance run", so an instance exists only for one task-instance run — never across
  runs, and never at registration.
- Testing a task is testing an ordinary method with a real return type whose injected arguments are
  passed in directly — no base class to subclass, no getters to override, no `In<T>` to unwrap.