Skip to content
4 changes: 3 additions & 1 deletion airflow-core/adr/lang-sdk/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,4 +35,6 @@ bind core interfaces and apply to every language SDK, not just the Java SDK.
- [ADR-0007](0007-taskflow-across-language-boundary.md): TaskFlow across the language boundary — argument binding for Lang-SDK tasks.

Decisions specific to a single SDK stay next to that SDK — for example, the Go SDK's bundle-format
decisions live in [`go-sdk/adr/`](../../../go-sdk/adr).
decisions live in [`go-sdk/adr/`](../../../go-sdk/adr). Java-SDK-only interface-design decisions —
the Java call-site syntax itself, rather than anything binding a core surface — live in
[`java-sdk/adr/`](../../../java-sdk/adr), distinct from the cross-cutting ADRs kept here.
200 changes: 200 additions & 0 deletions java-sdk/adr/0001-mixed-lang-dag-interface.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,200 @@
<!--
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-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`](../../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](0002-native-dag-interface.md)). 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

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

```java
// 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);
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

```java
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`

```java
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 bundle'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:

```java
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](0002-native-dag-interface.md)), 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` bundle, 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 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.
- Missing reference or boxed values become `null`; primitive inputs fail clearly.
- `@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](0002-native-dag-interface.md)). `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.
Loading