ADR-0001: Mixed-Lang Dag — Java Task Interface

Status

Proposed

Context

The Python @task.stub call site already defines task data flow. Java tasks should consume those bindings directly instead of repeating upstream ids with @Builder.XCom.

This ADR is scoped to the Java call-site interface only — the shape of the argument-binding spec itself, how it is materialized, and how it travels over the wire is the protocol-level decision recorded in airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md. This ADR only answers: given that spec, what does the Java code a user writes look like?

Decision

The annotation is @Builder.TaskHandler(dag, task), not @Builder.Task(id). Python declares the task with @task.stub and Java supplies only its body, so the two are different things and should not share a name: @Builder.Dag/@Builder.Task stay with the native surface, where Java owns the Dag (ADR-0002). Python owns both ids too, so a handler names the pair it binds to and the processor generates its registration from them.

We support 3 authoring syntaxes for the TaskFlow binding. All three resolve to the same internal TaskArgs, which is not part of the public surface: a Dag author never names it, and only generated code reads it.

1. Annotation based with positional injection

@Builder.TaskHandler(dag = "etl", task = "score")
public long score(Client client, long rows, double threshold, List<String> regions) { ... }

Data parameters bind by position, and Client/Context are injected without consuming a position. The processor compiles this into a Task:

// Generated by the Airflow Java SDK annotation processor. Do not edit.
public static final class Score implements Task {
  @Override
  public void execute(Context context, Client client) throws Exception {
    TaskArgs args = TaskArgs.of(context, client, 3);
    long rows = args.require(0, Long.class);
    double threshold = args.require(1, Double.class);
    List<String> regions = args.require(2, new TypeReference<List<String>>() {});
    client.setXCom(new ExampleTasks().score(client, rows, threshold, regions));
  }
}

Each local carries its parameter's declared type, and TaskArgs converts by position — including a generic type, which a cast cannot express.

2. Annotation based with explicit struct

public static class ReportInput implements TaskInput {
  @ArgName("run_label")
  public String runLabel;

  public long transformed;
}

@Builder.TaskHandler(dag = "etl", task = "report")
public void report(ReportInput input) {
  log.log(INFO, "Report {0} for transformed value {1}", input.runLabel, input.transformed);
  if (!"nightly".equals(input.runLabel)) {
    throw new RuntimeException("expected run label 'nightly' but got " + input.runLabel);
  }
}

3. Interface based with an explicit `TaskInput`

public static class SummarizeInput implements TaskInput {
    @ArgName("region_code")
    public String region;

    public long transformed;
  }

  // summarize(region_code=..., transformed=...) is called with keyword
  // arguments, which bind to the `TaskInput`'s fields by name.
  public static class Summarize implements InputTask<SummarizeInput> {
    public void execute(@NotNull Context context, Client client, SummarizeInput input) {
      log.log(
          INFO, "Summarize region {0} for transformed value {1}", input.region, input.transformed);
      if (!"emea".equals(input.region)) {
        throw new RuntimeException("expected region 'emea' but got " + input.region);
      }
    }
  }

Defining a TaskInput is the only way an interface task receives bound values. InputTask<TaskArgs> is not offered: TaskArgs is positional and untyped, which is right for generated code and wrong for code a person writes and later reads.

Registering and serving

A Bundle is what the server serves, and where handlers and Dags are registered — create it, register, serve:

public static void main(String[] args) {
  Bundle bundle = new Bundle()
      .register(ScoreTasks.class)                    // annotation based: ids from the annotations
      .register("etl", "summarize", Summarize.class) // interface based: ids at the call
      .register(EtlPipeline.class);                  // native Dags share the bundle

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

One verb, overloaded, and everything registers as a class: nothing is constructed here.

Consequences

One binding contract serves both the annotation and interface authoring surfaces, because all three syntaxes resolve to the same internal TaskArgs.
Keeping TaskArgs internal means positional access stays available where it is cheap and safe — in code the processor writes and type-checks — without asking a Dag author to track argument order by hand.
@Builder.XCom is retired; the Python call site is the only source of data-flow wiring.
Renaming the shipped @Builder.Task to @Builder.TaskHandler(dag, task) for this surface breaks existing mixed-language Dags, which change one annotation and name the Dag they belong to.
Client/Context are injected as method arguments. This follows the same rule the native surface settles (ADR-0002), so one injection model spans both surfaces.

Appendix: Implementation Notes

Annotated task data parameters bind by position; injected Client and Context do not consume positions. Positional binding also needs no parameter names in the class file, so it does not depend on compiling with -parameters.
Keyword arguments bind through one TaskInput, with @ArgName only when Python and Java names differ.
Interface tasks use InputTask<MyInput> for named fields. There is no public positional form.
TaskArgs is internal. It is what generated code reads, and what every syntax above resolves to; nothing in the public API returns or accepts it. It is opened with of(context, client, declared) — the client because resolving a binding may have to pull an upstream's XCom, and the declared count because positions carry the whole meaning of a flat binding, so a call site that bound a different number of arguments than the method takes has already shifted them. A parameter the call site omitted is exempt: its default still arrives, but a method that does not declare it is not reading shifted arguments, so those are dropped before the counts are compared. It exposes require(index, type), which fails when the position resolves to nothing, and get(index, type), which yields null — each with a Class<T> and a TypeReference<T> form.
Annotation processing generates code, it never rewrites it, so the method a user writes stays exactly as written and the generated Task delegates to it. That also sets the two constraints the processor enforces at compile time: the annotated method has to be reachable from the generated class in the same package, and its declaring class needs a no-arg constructor, since the generated body calls new Parent().method(...).
A parameter whose type has type arguments binds through the TypeReference<T> overload, so the element type survives and the generated code never contains an unchecked cast.
Neither direction of a TaskInput name mismatch fails the task. A field binds by name, so a field nothing supplies keeps its Java default and an argument no field claims changes nothing the task reads. Both are logged, since either one means the Java signature and the Python stub disagree. Captured defaults are not logged, since the Dag author never passed them. Flat positional binding still fails either way, because a dropped or added argument shifts every later one.
Two fields whose argument names fold alike fail when the bundle is built, because the fold cannot tell them apart and there is no value either could safely take. Two arguments that fold alike are the other way round: they reach a field naming one of them exactly, and a field that would match both is left unfilled and reported.
An argument that resolves to nothing is a value, not a mistake. Once a parameter has claimed its argument, a null literal or an upstream that pushed no XCom gives null to a reference or boxed type, and fails for a primitive, which cannot hold it. Declaring a boxed type is how a task says the value is optional.
This matches the Go SDK, which logs both directions and zero-values an unmatched struct field (go-sdk ADR-0006), and it follows the cross-language rule in lang-SDK ADR-0007.
@Builder.XCom is removed, keeping the Python call site as the single source of data-flow wiring.
The generated Task is one more branch of an existing classifier. BuilderProcessor already compiles annotated methods into a Task implementation, classifying each parameter by type and emitting new Parent().method(...); positional binding adds a case to that classifier rather than a mechanism.
Locals are emitted with the parameter's declared type, never var, so the generated file reads as the method signature it serves, and a binding that does not fit is a compile error in generated code rather than a surprise at run time.
TaskArgs converts rather than casts, through the ObjectMapper the SDK already uses. That is what makes a generic parameter work: a cast cannot express List<String>, and erasure would let (List<String>) value succeed over a list of numbers and fail much later at the first element read. A Class<T> overload covers simple types and a TypeReference<T> overload carries the full generic type, the processor choosing by whether the declared type has type arguments. Conversion also subsumes numeric coercion, which the XCom path handles today by widening through Number by hand, since the wire decodes integers to Long and floats to Double. require names the parameter it failed on, so a missing value for a primitive fails clearly instead of surfacing as an unboxing NullPointerException.
register is overloaded, not suffixed: register(Class<?>) for anything annotated, register(String, String, Class<? extends Task>) for an InputTask with no annotation to read, and register(Dag) for a Dag built through the interface surface (ADR-0002). register(ScoreTasks.class) registers every handler the class holds. It is this ADR's proposed addition: today's Bundle takes an Iterable<Dag> through BundleBuilder.getDags(), with no place for a handler that belongs to a Python-owned Dag.
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.

12 KiB Raw Permalink Blame History