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

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: 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.

Decision

Annotation based

Client and Context are injected as method arguments, exactly as on the mixed-language surface (ADR-0001) — 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:

@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>:

// 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:

// 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) — 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):

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:

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:

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

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:

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), 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), 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), 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) — 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). 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.