From 816f911ad8f5b5ecd93776a3c8d6c06c5e2ab43d Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 24 Aug 2026 07:18:13 +0000 Subject: [PATCH 1/8] Add Java SDK interface-design ADRs for mixed-lang and native Dags --- airflow-core/adr/lang-sdk/README.md | 4 +- java-sdk/adr/0001-mixed-lang-dag-interface.md | 128 +++++++++++++++ java-sdk/adr/0002-native-dag-interface.md | 154 ++++++++++++++++++ 3 files changed, 285 insertions(+), 1 deletion(-) create mode 100644 java-sdk/adr/0001-mixed-lang-dag-interface.md create mode 100644 java-sdk/adr/0002-native-dag-interface.md diff --git a/airflow-core/adr/lang-sdk/README.md b/airflow-core/adr/lang-sdk/README.md index ba7800a97407b..b6768d6139610 100644 --- a/airflow-core/adr/lang-sdk/README.md +++ b/airflow-core/adr/lang-sdk/README.md @@ -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. diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md new file mode 100644 index 0000000000000..650e98e6649fa --- /dev/null +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -0,0 +1,128 @@ + + +# 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 + +We support 4 syntaxes for the TaskFlow binding: + +### 1. Annotation based with positional injection + +```java +@Builder.Task(id = "score") +public long score(Client client, long rows, double threshold) { ... } +``` + +### 2. Annotation based with explicit struct + +```java +public static class ReportInput implements TaskInput { + @ArgName("run_label") + public String runLabel; + + public long transformed; +} + +@Builder.Task(id = "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 explicit struct + +```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 { + 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); + } + } + } +``` + +### 4. Interface based with `TaskArgs` getter + +```java +public static class Transform implements InputTask { + public void execute(@NotNull Context context, Client client, TaskArgs args) { + var extracted = args.require(0, Long.class); + var threshold = args.get(1, Double.class); // null when it resolves to nothing + log.log(INFO, "Got extracted value from the bound argument: {0}", extracted); + // ... +} +``` + +### How + +- Annotated task data parameters bind by position; injected `Client` and `Context` do not consume + positions. +- Keyword arguments bind through one `TaskInput` bundle, with `@ArgName` only when Python and Java + names differ. +- Interface tasks use `InputTask` for positional access or `InputTask` for named + fields. +- 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. + +## Open Questions + +- Not sure should we just keep the `TaskArgs` as internal purpose or it's fine to keep it as public + interface? +- Should we replace the injectable `client` and `context` with explicit getter method (e.g. + `getClient` or `getContext`) to respect the real argument type on static check? Currently, we + need to introduce a runtime class injected by `@Wiring` to workaround the type mismatch error. + (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the + native-Dag side.) + +## Consequences + +- One binding contract serves both the annotation and interface authoring surfaces. +- `@Builder.XCom` is retired; the Python call site is the only source of data-flow wiring. +- The public surface is not final until the two open questions above are resolved. diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md new file mode 100644 index 0000000000000..546b77a40f89f --- /dev/null +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -0,0 +1,154 @@ + + +# 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 shares the injectable +`client`/`context` question raised in [ADR-0001](0001-mixed-lang-dag-interface.md), and 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 + +```java +@Builder.Dag(id = "java_etl", schedule = "@daily") +public class EtlPipeline { + + @Builder.Task(id = "extract", retries = 2) + public long extract(Client client) { + return 42L; + } + + @Builder.Task(id = "transform") + public long transform(Client client, long extracted) { + return extracted * 2; + } + + @Builder.Task(id = "load") + public void load(Context ctx, long transformed) { + // ... + } + + @Wiring + static void depends(EtlPipelineRef f) { + f.load(f.transform(f.extract())); + } +} +``` + +`f` is a generated `EtlPipelineRef` twin: calling it registers the task, its data parameters +become `In` inputs, and its return value becomes a `TaskRef`. The call graph is the task +graph, and `javac` type-checks it. + +### Interface based + +```java +var extract = new TaskDef("extract", Extract.class).config("retries", 2); +var transform = new TaskDef("transform", Transform.class).dependsOn(extract); + +var dag = new DagDef("java_etl") + .config("schedule", "@daily") + .addTask(extract) + .addTask(transform); +``` + +`TaskDef`/`DagDef` carry configuration through fluent `.config(key, value)` calls and dependency +edges through `.dependsOn(...)`, keyed to the Dag serialization schema. + +## If `client`/`context` were real getter methods instead of injected parameters + +The `Ref` twin above exists for one reason: the real `@Builder.Task` method's signature +takes `Client`/`Context` parameters, so it cannot be called directly inside `@Wiring` without a +second, type-compatible surface. If `client`/`context` are getter methods instead — not method +parameters — a task method's parameter list becomes data-only, and the real method is what +`@Wiring` calls. No twin, no `In`/`TaskRef` indirection: + +```java +// Before: @Wiring is static and calls a generated twin, because extract()/ +// transform() above take a Client parameter the twin has to drop. +@Wiring +static void depends(EtlPipelineRef f) { + f.load(f.transform(f.extract())); +} +``` + +```java +// After: client/context come from getters, so extract()/transform()/load() +// are data-only and directly composable. @Wiring calls the real methods. +@Builder.Dag(id = "java_etl", schedule = "@daily") +public class EtlPipeline extends Task { + + @Builder.Task(id = "extract") + public long extract() { + var client = getClient(); + return 42L; + } + + @Builder.Task(id = "transform") + public long transform(long extracted) { + return extracted * 2; + } + + @Builder.Task(id = "load") + public void load(long transformed) { + // ... + } + + @Wiring + void depends() { + load(transform(extract())); + } +} +``` + +`extract()`, `transform()`, and `load()` are the same methods `@Builder.Task` marks as task +bodies. Once `Client`/`Context` are getters, their real signatures already line up +(`long -> long -> void`), so this type-checks with no generated class produced purely to make the +types match. + +Open tradeoffs this ADR does not resolve: + +- What calling `extract()` inside `depends()` actually does — whether Dag parsing runs against a + distinct instance/mode so the call records an edge instead of executing the real task body — is + a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. +- A task method becomes an ordinary instance method with a real return type, so it is directly + unit-testable by subclassing and overriding `getClient()`/`getContext()` — no `In` unwrapping + needed in a test. +- The interface-based `TaskDef`/`DagDef` surface is unaffected either way: it already wires edges + through object references (`dependsOn(extract)`), not through compile-time-generated calls. +- If adopted, the annotation surface drops `In`, `TaskRef`, and `Ref` codegen + entirely, along with the "IDE shows unresolved symbols until the first build" experience that + comes with a generated twin. + +## Consequences + +- Two authoring surfaces (annotation, interface) both flow into the same `DagDef`/`TaskDef` model. +- The annotation surface's ergonomics are still open: keeping `@Wiring` + a twin is the safer, + already-scoped design; dropping it for getters is a real simplification but needs the + wiring-time-vs-run-time dispatch question answered before it can be adopted. From 1b9537649dc29129b9bf283e45d0a20cc760df5b Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 24 Aug 2026 07:27:45 +0000 Subject: [PATCH 2/8] Address review nits in Java SDK interface-design ADRs --- java-sdk/adr/0001-mixed-lang-dag-interface.md | 11 ++++++----- java-sdk/adr/0002-native-dag-interface.md | 15 ++++++++++----- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md index 650e98e6649fa..0ff117f8bbad2 100644 --- a/java-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -91,11 +91,12 @@ public static class SummarizeInput implements TaskInput { ```java public static class Transform implements InputTask { - public void execute(@NotNull Context context, Client client, TaskArgs args) { - var extracted = args.require(0, Long.class); - var threshold = args.get(1, Double.class); // null when it resolves to nothing - log.log(INFO, "Got extracted value from the bound argument: {0}", extracted); - // ... + public void execute(@NotNull Context context, Client client, TaskArgs args) { + var extracted = args.require(0, Long.class); + var threshold = args.get(1, Double.class); // null when it resolves to nothing + log.log(INFO, "Got extracted value from the bound argument: {0}", extracted); + // ... + } } ``` diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index 546b77a40f89f..6ce98fdb8667d 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -79,7 +79,9 @@ var dag = new DagDef("java_etl") ``` `TaskDef`/`DagDef` carry configuration through fluent `.config(key, value)` calls and dependency -edges through `.dependsOn(...)`, keyed to the Dag serialization schema. +edges through `.dependsOn(...)`, keyed to the Dag serialization schema. `.config(...)` and +`.dependsOn(...)` are the native-Dag additions this ADR proposes for `TaskDef`/`DagDef`; today's +shipped classes only have `addTask(...)`. ## If `client`/`context` were real getter methods instead of injected parameters @@ -132,19 +134,22 @@ bodies. Once `Client`/`Context` are getters, their real signatures already line (`long -> long -> void`), so this type-checks with no generated class produced purely to make the types match. -Open tradeoffs this ADR does not resolve: +Open question this ADR does not resolve: - What calling `extract()` inside `depends()` actually does — whether Dag parsing runs against a distinct instance/mode so the call records an edge instead of executing the real task body — is a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. + +If that question is answered, the implications are: + - A task method becomes an ordinary instance method with a real return type, so it is directly unit-testable by subclassing and overriding `getClient()`/`getContext()` — no `In` unwrapping needed in a test. - The interface-based `TaskDef`/`DagDef` surface is unaffected either way: it already wires edges through object references (`dependsOn(extract)`), not through compile-time-generated calls. -- If adopted, the annotation surface drops `In`, `TaskRef`, and `Ref` codegen - entirely, along with the "IDE shows unresolved symbols until the first build" experience that - comes with a generated twin. +- The annotation surface drops `In`, `TaskRef`, and `Ref` codegen entirely, along with + the "IDE shows unresolved symbols until the first build" experience that comes with a generated + twin. ## Consequences From 81c35cf0b284db4da1b0e5f38d50cb91c6d3e066 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 07:46:06 +0000 Subject: [PATCH 3/8] Record the reviewed Java SDK stub handler and native Dag interfaces Review on #72019 asked for TaskArgs to leave the public surface, and following that through reshaped more than one interface. Positional, untyped access is right for code a processor emits and wrong for code a person writes and later reads, so an interface task now binds through a TaskInput, and TaskArgs is reached only from generated code. The native Dag interface changed for a different reason. Its wiring expression composed a generated twin rather than the task methods themselves, so a Dag author read one set of signatures and ran another. Making Client and Context getters removes that mismatch at the source: the methods the wiring composes are the methods that run, and javac checks them against each other with nothing generated in between. An annotation also no longer calls a mixed-language body a task. Python declares that task and Java supplies only its body, which is what @Builder.StubHandler names. --- java-sdk/adr/0001-mixed-lang-dag-interface.md | 136 +++++++++--- java-sdk/adr/0002-native-dag-interface.md | 204 +++++++++++------- 2 files changed, 229 insertions(+), 111 deletions(-) diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md index 0ff117f8bbad2..5ff1eccf97a76 100644 --- a/java-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -36,15 +36,43 @@ This ADR only answers: given that spec, what does the Java code a user writes lo ## Decision -We support 4 syntaxes for the TaskFlow binding: +The annotation is `@Builder.StubHandler(dagId, taskId)`, 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.Task(id = "score") -public long score(Client client, long rows, double threshold) { ... } +@Builder.StubHandler(dagId = "etl", taskId = "score") +public long score(Client client, long rows, double threshold, List 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 regions = args.require(2, new TypeReference>() {}); + 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 @@ -55,7 +83,7 @@ public static class ReportInput implements TaskInput { public long transformed; } -@Builder.Task(id = "report") +@Builder.StubHandler(dagId = "etl", taskId = "report") public void report(ReportInput input) { log.log(INFO, "Report {0} for transformed value {1}", input.runLabel, input.transformed); if (!"nightly".equals(input.runLabel)) { @@ -64,7 +92,7 @@ public void report(ReportInput input) { } ``` -### 3. Interface based with explicit struct +### 3. Interface based with an explicit `TaskInput` ```java public static class SummarizeInput implements TaskInput { @@ -87,43 +115,85 @@ public static class SummarizeInput implements TaskInput { } ``` -### 4. Interface based with `TaskArgs` getter +Defining a `TaskInput` is the only way an interface task receives bound values. `InputTask` +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 class Transform implements InputTask { - public void execute(@NotNull Context context, Client client, TaskArgs args) { - var extracted = args.require(0, Long.class); - var threshold = args.get(1, Double.class); // null when it resolves to nothing - log.log(INFO, "Got extracted value from the bound argument: {0}", extracted); - // ... - } +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); } ``` -### How +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.StubHandler(dagId, taskId)` for this surface + breaks existing mixed-language Dags, which change one annotation and name the Dag they belong to. +- The public surface is not final until the open question above is resolved. + +## Appendix: Implementation Notes - Annotated task data parameters bind by position; injected `Client` and `Context` do not consume - positions. + 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` for positional access or `InputTask` for named - fields. +- Interface tasks use `InputTask` 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` and a `TypeReference` 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` 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. - -## Open Questions - -- Not sure should we just keep the `TaskArgs` as internal purpose or it's fine to keep it as public - interface? -- Should we replace the injectable `client` and `context` with explicit getter method (e.g. - `getClient` or `getContext`) to respect the real argument type on static check? Currently, we - need to introduce a runtime class injected by `@Wiring` to workaround the type mismatch error. - (See [ADR-0002](0002-native-dag-interface.md) for where this same question resurfaces on the - native-Dag side.) - -## Consequences - -- One binding contract serves both the annotation and interface authoring surfaces. -- `@Builder.XCom` is retired; the Python call site is the only source of data-flow wiring. -- The public surface is not final until the two open questions above are resolved. +- **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`, and erasure would let + `(List) value` succeed over a list of numbers and fail much later at the first element + read. A `Class` overload covers simple types and a `TypeReference` 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)` 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` + 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)` 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. diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index 6ce98fdb8667d..2f51932e428b7 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -36,124 +36,172 @@ protocol substrate (the argument-binding spec) with ### Annotation based +A task method takes only its data. `Client` and `Context` come from getters on `Tasks`, the SDK base +class, so each method's real signature is exactly what the wiring expression composes: + ```java @Builder.Dag(id = "java_etl", schedule = "@daily") -public class EtlPipeline { +public class EtlPipeline extends Tasks { @Builder.Task(id = "extract", retries = 2) - public long extract(Client client) { - return 42L; + public long extract() { + return ((Number) getClient().getVariable("row_count")).longValue(); } @Builder.Task(id = "transform") - public long transform(Client client, long extracted) { + public long transform(long extracted) { + getClient().setXCom("scaled_in", getContext().runId); return extracted * 2; } @Builder.Task(id = "load") - public void load(Context ctx, long transformed) { - // ... + public void load(long transformed) { + log.log(INFO, "Loaded {0} for run {1}", transformed, getContext().runId); } @Wiring - static void depends(EtlPipelineRef f) { - f.load(f.transform(f.extract())); + void depends() { + load(transform(extract())); + } + + public static void main(String[] args) { + Bundle bundle = new Bundle().register(EtlPipeline.class); + Server.create(args).serve(bundle); } } ``` -`f` is a generated `EtlPipelineRef` twin: calling it registers the task, its data parameters -become `In` inputs, and its return value becomes a `TaskRef`. The call graph is the task -graph, and `javac` type-checks it. +`javac` type-checks `load(transform(extract()))` against the methods that will actually run, because +`long -> long -> void` is already their real shape. `@Wiring` is read, never called: the processor +turns the call expression into Dag edges, and no task body runs at Dag-parse time. -### Interface based +The result is the Dag below, equivalent to writing the interface-based form by hand: ```java -var extract = new TaskDef("extract", Extract.class).config("retries", 2); -var transform = new TaskDef("transform", Transform.class).dependsOn(extract); +// 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).after(extract); + dag.task("load", Load.class).after(transform); + return dag; + } -var dag = new DagDef("java_etl") - .config("schedule", "@daily") - .addTask(extract) - .addTask(transform); -``` + public static final class Transform implements Task { + @Override + public void execute(Context context, Client client) throws Exception { + TaskArgs args = TaskArgs.of(context); + long extracted = args.require(0, Long.class); + EtlPipeline tasks = new EtlPipeline(); + tasks.bind(context, client); + client.setXCom(tasks.transform(extracted)); + } + } -`TaskDef`/`DagDef` carry configuration through fluent `.config(key, value)` calls and dependency -edges through `.dependsOn(...)`, keyed to the Dag serialization schema. `.config(...)` and -`.dependsOn(...)` are the native-Dag additions this ADR proposes for `TaskDef`/`DagDef`; today's -shipped classes only have `addTask(...)`. + // Extract and Load follow the same shape. +} +``` -## If `client`/`context` were real getter methods instead of injected parameters +`bind(context, client)` is what `getClient()` and `getContext()` return for that invocation, and a +getter called with nothing bound throws. -The `Ref` twin above exists for one reason: the real `@Builder.Task` method's signature -takes `Client`/`Context` parameters, so it cannot be called directly inside `@Wiring` without a -second, type-compatible surface. If `client`/`context` are getter methods instead — not method -parameters — a task method's parameter list becomes data-only, and the real method is what -`@Wiring` calls. No twin, no `In`/`TaskRef` indirection: +### Interface based ```java -// Before: @Wiring is static and calls a generated twin, because extract()/ -// transform() above take a Client parameter the twin has to drop. -@Wiring -static void depends(EtlPipelineRef f) { - f.load(f.transform(f.extract())); -} +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).after(extract); +TaskRef load = dag.task("load", Load.class).after(transform); + +load.before(dag.task("notify", Notify.class)); ``` -```java -// After: client/context come from getters, so extract()/transform()/load() -// are data-only and directly composable. @Wiring calls the real methods. -@Builder.Dag(id = "java_etl", schedule = "@daily") -public class EtlPipeline extends Task { +`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. `before` and `after` are the edge verbs, variadic and mirror +images of each other — Java's spelling of Python's `>>` and `<<` — and `.config(key, value)` carries +Dag and task configuration. - @Builder.Task(id = "extract") - public long extract() { - var client = getClient(); - return 42L; - } +### Registering and serving - @Builder.Task(id = "transform") - public long transform(long extracted) { - return extracted * 2; - } +Both surfaces produce a `Dag`, and a `Bundle` is what the server serves — create it, register, serve: - @Builder.Task(id = "load") - public void load(long transformed) { - // ... - } +```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 - @Wiring - void depends() { - load(transform(extract())); - } + Server.create(args).serve(bundle); } ``` -`extract()`, `transform()`, and `load()` are the same methods `@Builder.Task` marks as task -bodies. Once `Client`/`Context` are getters, their real signatures already line up -(`long -> long -> void`), so this type-checks with no generated class produced purely to make the -types match. +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. -Open question this ADR does not resolve: +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. -- What calling `extract()` inside `depends()` actually does — whether Dag parsing runs against a - distinct instance/mode so the call records an edge instead of executing the real task body — is - a runtime-dispatch question, not a call-site question. This ADR only proposes the shape. +One bundle carries native Dags and mixed-language stub handlers alike +([ADR-0001](0001-mixed-lang-dag-interface.md)), so one process serves both. -If that question is answered, the implications are: +## Alternatives -- A task method becomes an ordinary instance method with a real return type, so it is directly - unit-testable by subclassing and overriding `getClient()`/`getContext()` — no `In` unwrapping - needed in a test. -- The interface-based `TaskDef`/`DagDef` surface is unaffected either way: it already wires edges - through object references (`dependsOn(extract)`), not through compile-time-generated calls. -- The annotation surface drops `In`, `TaskRef`, and `Ref` codegen entirely, along with - the "IDE shows unresolved symbols until the first build" experience that comes with a generated - twin. +- **Keeping `Client`/`Context` as task parameters.** The wiring expression would then compose + different signatures than the ones that run, so a bridge has to exist somewhere: either the wiring + names the injectables (`load(ctx, transform(client, ctx, extract(client)))`), or the user extends a + generated supertype holding data-only overloads — a name in user source that an IDE cannot resolve + until the first build. Getters remove the mismatch instead of bridging it. +- **Capturing the graph at run time** by generating a subclass that overrides each task + method to record an edge and return a placeholder. Virtual dispatch makes the inherited `depends()` + call the overrides, so no task body runs — but a `long` return carries no identity, so a result + stored in a local and used twice cannot be attributed to its producer. Reading the tree has no + such blind spot. ## Consequences -- Two authoring surfaces (annotation, interface) both flow into the same `DagDef`/`TaskDef` model. -- The annotation surface's ergonomics are still open: keeping `@Wiring` + a twin is the safer, - already-scoped design; dropping it for getters is a real simplification but needs the - wiring-time-vs-run-time dispatch question answered before it can be adopted. +- Two authoring surfaces (annotation, interface) both flow into the same `Dag`/`TaskRef` model, and + the annotation surface generates exactly the interface-based calls a user could have written. +- The annotation surface generates only what runs — a Dag definition and one `Task` per method — and + nothing whose only purpose is to make user code compile. +- A task method is plain Java with a real signature, so nothing about the authoring surface exists + only to satisfy the compiler: no twin class, no `In`, no `TaskRef` wrapper. +- Native tasks keep `@Builder.Task`, while the mixed-language surface moves to + `@Builder.StubHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names + which of the two things a method is. +- The interface-based surface wires edges through object references (`after(extract)`) rather than + through compile-time-generated calls, so it needs no tree reading and works under any compiler. + +## Appendix: Implementation Notes + +- The processor walks the `@Wiring` body's syntax tree and emits one `after` edge per nested call, + ignoring arguments whose static type is `Client` or `Context`, so the expression above becomes + `extract -> transform -> load`. A task result held in a local and passed to two downstream tasks is + ordinary tree analysis, and a call that is not a task method is a compile error reported on its own + source line through `Messager`. +- `com.sun.source.util.Trees` is exported by `jdk.compiler` and is stable to use from a processor, + but it is javac's tree model: a processor built on it does not run under a compiler with its own + model, such as ecj. Gradle drives javac here, so this is a constraint to record rather than one to + solve. +- A task method is an ordinary method with a real return type, so a test subclasses the Dag class and + overrides `getClient()`/`getContext()`, with no `In` to unwrap. +- Resolving a task result held in a local is required, not optional. `long e = extract(); fanin(e, ping());` + reads as an identifier rather than a call, and a walker that only follows nested invocations drops + the `extract -> fanin` edge **silently**. The processor resolves an identifier through the local's + initializer, and rejects an argument it cannot attribute rather than emitting an incomplete graph. +- **The getters need no thread-local.** `bind` sets a field on `Tasks`, and the generated code already + constructs the instance, so nothing ambient is required — which matters because the SDK's toolchain + is pinned to Java 11, where `ScopedValue` does not exist. +- **`dag.task(...)` is a factory method, not a constructor.** Java has no `new dag.TaskRef(...)`: + qualified inner-class creation is spelled `dag.new TaskRef(...)`, which reads as compiler trivia + rather than as a Dag definition. +- **`dag.task(...)`, `.config(...)`, `.before(...)`, and `.after(...)` 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. +- **`register` is one overloaded verb**: `register(Class)` for an annotated class, `register(Dag)` + for a built one, and `register(dagId, taskId, Class)` for a stub handler + ([ADR-0001](0001-mixed-lang-dag-interface.md)). It is proposed here; today a bundle is built from an + `Iterable` through `BundleBuilder.getDags()`. From 9d6d7efa5e3231fa7ab4fb9ef44ea2c4b18b141b Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 08:38:59 +0000 Subject: [PATCH 4/8] Rename StubHandler to StubTask in the Java SDK ADRs The name should say what the SDK registers, and what it registers is a task: Python declares it with @task.stub, gives it a task_id, and the SDK supplies only its body. "Handler" described the callable rather than the thing it stands for, leaving these docs using a word no other part of Airflow uses for a task. --- java-sdk/adr/0001-mixed-lang-dag-interface.md | 8 ++++---- java-sdk/adr/0002-native-dag-interface.md | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md index 5ff1eccf97a76..d3d0a0029d25f 100644 --- a/java-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -36,7 +36,7 @@ This ADR only answers: given that spec, what does the Java code a user writes lo ## Decision -The annotation is `@Builder.StubHandler(dagId, taskId)`, not `@Builder.Task(id)`. Python declares the +The annotation is `@Builder.StubTask(dagId, taskId)`, 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 @@ -49,7 +49,7 @@ code reads it. ### 1. Annotation based with positional injection ```java -@Builder.StubHandler(dagId = "etl", taskId = "score") +@Builder.StubTask(dagId = "etl", taskId = "score") public long score(Client client, long rows, double threshold, List regions) { ... } ``` @@ -83,7 +83,7 @@ public static class ReportInput implements TaskInput { public long transformed; } -@Builder.StubHandler(dagId = "etl", taskId = "report") +@Builder.StubTask(dagId = "etl", taskId = "report") public void report(ReportInput input) { log.log(INFO, "Report {0} for transformed value {1}", input.runLabel, input.transformed); if (!"nightly".equals(input.runLabel)) { @@ -145,7 +145,7 @@ One verb, overloaded, and everything registers as a class: nothing is constructe 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.StubHandler(dagId, taskId)` for this surface +- Renaming the shipped `@Builder.Task` to `@Builder.StubTask(dagId, taskId)` for this surface breaks existing mixed-language Dags, which change one annotation and name the Dag they belong to. - The public surface is not final until the open question above is resolved. diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index 2f51932e428b7..a037ff6c7254a 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -145,7 +145,7 @@ Nothing is constructed at registration. Task classes are instantiated per task-i 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 stub handlers alike +One bundle carries native Dags and mixed-language stub tasks alike ([ADR-0001](0001-mixed-lang-dag-interface.md)), so one process serves both. ## Alternatives @@ -170,7 +170,7 @@ One bundle carries native Dags and mixed-language stub handlers alike - A task method is plain Java with a real signature, so nothing about the authoring surface exists only to satisfy the compiler: no twin class, no `In`, no `TaskRef` wrapper. - Native tasks keep `@Builder.Task`, while the mixed-language surface moves to - `@Builder.StubHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names + `@Builder.StubTask` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names which of the two things a method is. - The interface-based surface wires edges through object references (`after(extract)`) rather than through compile-time-generated calls, so it needs no tree reading and works under any compiler. @@ -202,6 +202,6 @@ One bundle carries native Dags and mixed-language stub handlers alike 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. - **`register` is one overloaded verb**: `register(Class)` for an annotated class, `register(Dag)` - for a built one, and `register(dagId, taskId, Class)` for a stub handler + for a built one, and `register(dagId, taskId, Class)` for a stub task ([ADR-0001](0001-mixed-lang-dag-interface.md)). It is proposed here; today a bundle is built from an `Iterable` through `BundleBuilder.getDags()`. From 78265ec70a1cc8fb7fb981c6e5544eeb40e698d7 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Wed, 9 Sep 2026 10:53:42 +0000 Subject: [PATCH 5/8] Rename StubTask to TaskHandler in the Java SDK ADRs Python declares the task; an SDK supplies only its body. Naming the SDK object after the task overloads a word the native surface already uses for something the user creates with dag.Task, so it names the role the SDK actually plays instead. --- java-sdk/adr/0001-mixed-lang-dag-interface.md | 8 ++++---- java-sdk/adr/0002-native-dag-interface.md | 6 +++--- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md index d3d0a0029d25f..e30ebdc95d65a 100644 --- a/java-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -36,7 +36,7 @@ This ADR only answers: given that spec, what does the Java code a user writes lo ## Decision -The annotation is `@Builder.StubTask(dagId, taskId)`, not `@Builder.Task(id)`. Python declares the +The annotation is `@Builder.TaskHandler(dagId, taskId)`, 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 @@ -49,7 +49,7 @@ code reads it. ### 1. Annotation based with positional injection ```java -@Builder.StubTask(dagId = "etl", taskId = "score") +@Builder.TaskHandler(dagId = "etl", taskId = "score") public long score(Client client, long rows, double threshold, List regions) { ... } ``` @@ -83,7 +83,7 @@ public static class ReportInput implements TaskInput { public long transformed; } -@Builder.StubTask(dagId = "etl", taskId = "report") +@Builder.TaskHandler(dagId = "etl", taskId = "report") public void report(ReportInput input) { log.log(INFO, "Report {0} for transformed value {1}", input.runLabel, input.transformed); if (!"nightly".equals(input.runLabel)) { @@ -145,7 +145,7 @@ One verb, overloaded, and everything registers as a class: nothing is constructe 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.StubTask(dagId, taskId)` for this surface +- Renaming the shipped `@Builder.Task` to `@Builder.TaskHandler(dagId, taskId)` for this surface breaks existing mixed-language Dags, which change one annotation and name the Dag they belong to. - The public surface is not final until the open question above is resolved. diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index a037ff6c7254a..7469a98dcfdc4 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -145,7 +145,7 @@ Nothing is constructed at registration. Task classes are instantiated per task-i 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 stub tasks alike +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 @@ -170,7 +170,7 @@ One bundle carries native Dags and mixed-language stub tasks alike - A task method is plain Java with a real signature, so nothing about the authoring surface exists only to satisfy the compiler: no twin class, no `In`, no `TaskRef` wrapper. - Native tasks keep `@Builder.Task`, while the mixed-language surface moves to - `@Builder.StubTask` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names + `@Builder.TaskHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names which of the two things a method is. - The interface-based surface wires edges through object references (`after(extract)`) rather than through compile-time-generated calls, so it needs no tree reading and works under any compiler. @@ -202,6 +202,6 @@ One bundle carries native Dags and mixed-language stub tasks alike 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. - **`register` is one overloaded verb**: `register(Class)` for an annotated class, `register(Dag)` - for a built one, and `register(dagId, taskId, Class)` for a stub task + for a built one, and `register(dagId, taskId, Class)` for a task handler ([ADR-0001](0001-mixed-lang-dag-interface.md)). It is proposed here; today a bundle is built from an `Iterable` through `BundleBuilder.getDags()`. From 1f7969eb709bdec4ad6aba193be7ee508dc762f8 Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Mon, 14 Sep 2026 18:29:24 +0800 Subject: [PATCH 6/8] Inject client/context as args in native Dag sytax The native Java Dag interface previously took Client and Context from getters on a Tasks base class. Switch to injecting them as method arguments, matching the mixed-language surface, so one injection rule spans the SDK, no ambient getter exists, and the Dag class keeps its single inheritance slot free. ADR-0001 (mixed-lang): - Flatten the @Builder.TaskHandler parameters from dagId/taskId to dag/task, and note client/context are injected as arguments, consistent with the native surface. ADR-0002 (native): - Wiring moves to a nested @Builder.Deps class implementing a generated Deps wiring-view interface: injected args stripped, data args as Arg, returns as TaskRef. - depends() is executed once in recording mode rather than read as a syntax tree, so it needs no com.sun.source.util.Trees and works under any compiler; a TaskRef carries node identity, so a result reused in a local just works. - Ordering-only ("non-TaskFlow") edges use then/after on Chain plus Flow.of; literal data args use lit(...). The chain/fanOut/fanIn shorthands are dropped as redundant with the two verbs. - Getters, syntax-tree reading, and subclass capture are recorded under Alternatives with their rejection rationale. The interface-based surface and the Bundle/register/Server flow are kept. --- java-sdk/adr/0001-mixed-lang-dag-interface.md | 11 +- java-sdk/adr/0002-native-dag-interface.md | 275 +++++++++++++----- 2 files changed, 201 insertions(+), 85 deletions(-) diff --git a/java-sdk/adr/0001-mixed-lang-dag-interface.md b/java-sdk/adr/0001-mixed-lang-dag-interface.md index e30ebdc95d65a..305f9c7df6ac5 100644 --- a/java-sdk/adr/0001-mixed-lang-dag-interface.md +++ b/java-sdk/adr/0001-mixed-lang-dag-interface.md @@ -36,7 +36,7 @@ This ADR only answers: given that spec, what does the Java code a user writes lo ## Decision -The annotation is `@Builder.TaskHandler(dagId, taskId)`, not `@Builder.Task(id)`. Python declares the +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 @@ -49,7 +49,7 @@ code reads it. ### 1. Annotation based with positional injection ```java -@Builder.TaskHandler(dagId = "etl", taskId = "score") +@Builder.TaskHandler(dag = "etl", task = "score") public long score(Client client, long rows, double threshold, List regions) { ... } ``` @@ -83,7 +83,7 @@ public static class ReportInput implements TaskInput { public long transformed; } -@Builder.TaskHandler(dagId = "etl", taskId = "report") +@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)) { @@ -145,9 +145,10 @@ One verb, overloaded, and everything registers as a class: nothing is constructe 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(dagId, taskId)` for this surface +- 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. -- The public surface is not final until the open question above is resolved. +- `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 diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index 7469a98dcfdc4..1b041fee02495 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -27,41 +27,58 @@ Proposed 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 shares the injectable -`client`/`context` question raised in [ADR-0001](0001-mixed-lang-dag-interface.md), and shares the -protocol substrate (the argument-binding spec) with +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 -A task method takes only its data. `Client` and `Context` come from getters on `Tasks`, the SDK base -class, so each method's real signature is exactly what the wiring expression composes: +`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 Tasks { +public class EtlPipeline { // extends nothing of ours; your own base class stays free @Builder.Task(id = "extract", retries = 2) - public long extract() { - return ((Number) getClient().getVariable("row_count")).longValue(); + public long extract(Client client) { + return ((Number) client.getVariable("row_count")).longValue(); } @Builder.Task(id = "transform") - public long transform(long extracted) { - getClient().setXCom("scaled_in", getContext().runId); - return extracted * 2; + 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(long transformed) { - log.log(INFO, "Loaded {0} for run {1}", transformed, getContext().runId); + public void load(Context context, long transformed) { + log.log(INFO, "Loaded {0} for run {1}", transformed, context.runId); } - @Wiring - void depends() { - load(transform(extract())); + @Builder.Task(id = "audit") + public void audit(Client client) { /* side effect only, no data in or out */ } + + @Builder.Task(id = "notify") + public void notify(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 + notify().after(loaded, audit()); // [load, audit] >> notify + } } public static void main(String[] args) { @@ -71,11 +88,36 @@ public class EtlPipeline extends Tasks { } ``` -`javac` type-checks `load(transform(extract()))` against the methods that will actually run, because -`long -> long -> void` is already their real shape. `@Wiring` is read, never called: the processor -turns the call expression into Dag edges, and no task body runs at Dag-parse time. +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 `Deps` (here +`EtlPipelineDeps`), one default method per task with the injected arguments stripped, each data +argument lifted to `Arg`, and the return lifted to `TaskRef`: -The result is the Dag below, equivalent to writing the interface-based form by hand: +```java +// Generated by the Airflow Java SDK annotation processor. Do not edit. +interface EtlPipelineDeps { + default TaskRef extract() { return Flow.node("extract"); } + default TaskRef transform(Arg extracted, Arg threshold) { return Flow.call("transform", extracted, threshold); } + default TaskRef load(Arg transformed) { return Flow.call("load", transformed); } + default TaskRef audit() { return Flow.node("audit"); } + default TaskRef notify() { 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`, which is an `Arg` 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. @@ -85,7 +127,9 @@ public final class EtlPipeline_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).after(extract); - dag.task("load", Load.class).after(transform); + TaskRef load = dag.task("load", Load.class).after(transform); + TaskRef audit = dag.task("audit", Audit.class).after(extract); // ordering-only edge + dag.task("notify", Notify.class).after(load, audit); // ordering-only edge return dag; } @@ -94,18 +138,57 @@ public final class EtlPipeline_Dag { public void execute(Context context, Client client) throws Exception { TaskArgs args = TaskArgs.of(context); long extracted = args.require(0, Long.class); - EtlPipeline tasks = new EtlPipeline(); - tasks.bind(context, client); - client.setXCom(tasks.transform(extracted)); + double threshold = 0.9; // baked from lit(0.9) at Dag-build time + client.setXCom(new EtlPipeline().transform(client, context, extracted, threshold)); } } - // Extract and Load follow the same shape. + // Extract, Load, Audit, and Notify follow the same shape. } ``` -`bind(context, client)` is what `getClient()` and `getContext()` return for that invocation, and a -getter called with nothing bound throws. +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`, 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-TaskFlow dependencies + +A TaskFlow (data) edge comes for free from passing a `TaskRef` into another wiring method. A +dependency where **no data flows** — Python's `a >> b`, and the list forms `a >> [b, c]` and +`[x, y] >> z` — is expressed instead with two variadic verbs that every `TaskRef` carries, `then` +and `after`. Both live on a small `Chain` interface, and a `TaskRef` is a `Chain` of one +(`interface TaskRef extends Arg, Chain`): + +```java +a.then(b, c); // a >> [b, c] +z.after(x, y); // z << [x, y] +``` + +`then` and `after` are mirror images, and each returns the **new frontier** (the set it just pointed +at), the way `>>`/`<<` evaluate to their 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 the verbs 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] +``` + +That is the whole non-TaskFlow surface: `then` and `after` on `Chain`, plus `Flow.of`. There is no +`chain`, `fanOut`, or `fanIn` — `a.then(b).then(c)` is a chain, `a.then(b, c)` is a fan-out, and +`z.after(x, y)` is a fan-in — so the shorthands would earn nothing. +Both edge kinds compile to the same `after(...)` 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 @@ -116,13 +199,15 @@ TaskRef extract = dag.task("extract", Extract.class).config("retries", 2); TaskRef transform = dag.task("transform", Transform.class).after(extract); TaskRef load = dag.task("load", Load.class).after(transform); -load.before(dag.task("notify", Notify.class)); +load.then(dag.task("notify", Notify.class)); ``` `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. `before` and `after` are the edge verbs, variadic and mirror -images of each other — Java's spelling of Python's `>>` and `<<` — and `.config(key, value)` carries -Dag and task configuration. +second `addTask(...)` call to forget. `then` and `after` are the same variadic edge verbs as the +annotation surface — Java's spelling of Python's `>>` and `<<` — 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 @@ -131,8 +216,8 @@ Both surfaces produce a `Dag`, and a `Bundle` is what the server serves — crea ```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 + .register(EtlPipeline.class) // annotation based: the user's own annotated class + .register(dag); // interface based: the Dag built above Server.create(args).serve(bundle); } @@ -150,58 +235,88 @@ One bundle carries native Dags and mixed-language task handlers alike ## Alternatives -- **Keeping `Client`/`Context` as task parameters.** The wiring expression would then compose - different signatures than the ones that run, so a bridge has to exist somewhere: either the wiring - names the injectables (`load(ctx, transform(client, ctx, extract(client)))`), or the user extends a - generated supertype holding data-only overloads — a name in user source that an IDE cannot resolve - until the first build. Getters remove the mismatch instead of bridging it. -- **Capturing the graph at run time** by generating a subclass that overrides each task - method to record an edge and return a placeholder. Virtual dispatch makes the inherited `depends()` - call the overrides, so no task body runs — but a `long` return carries no identity, so a result - stored in a local and used twice cannot be attributed to its producer. Reading the tree has no - such blind spot. +- **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. ## Consequences -- Two authoring surfaces (annotation, interface) both flow into the same `Dag`/`TaskRef` model, and - the annotation surface generates exactly the interface-based calls a user could have written. -- The annotation surface generates only what runs — a Dag definition and one `Task` per method — and - nothing whose only purpose is to make user code compile. -- A task method is plain Java with a real signature, so nothing about the authoring surface exists - only to satisfy the compiler: no twin class, no `In`, no `TaskRef` wrapper. -- Native tasks keep `@Builder.Task`, while the mixed-language surface moves to - `@Builder.TaskHandler` ([ADR-0001](0001-mixed-lang-dag-interface.md)), so an annotation names - which of the two things a method is. -- The interface-based surface wires edges through object references (`after(extract)`) rather than - through compile-time-generated calls, so it needs no tree reading and works under any compiler. +- 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 (`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`, 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 walks the `@Wiring` body's syntax tree and emits one `after` edge per nested call, - ignoring arguments whose static type is `Client` or `Context`, so the expression above becomes - `extract -> transform -> load`. A task result held in a local and passed to two downstream tasks is - ordinary tree analysis, and a call that is not a task method is a compile error reported on its own - source line through `Messager`. -- `com.sun.source.util.Trees` is exported by `jdk.compiler` and is stable to use from a processor, - but it is javac's tree model: a processor built on it does not run under a compiler with its own - model, such as ecj. Gradle drives javac here, so this is a constraint to record rather than one to - solve. -- A task method is an ordinary method with a real return type, so a test subclasses the Dag class and - overrides `getClient()`/`getContext()`, with no `In` to unwrap. -- Resolving a task result held in a local is required, not optional. `long e = extract(); fanin(e, ping());` - reads as an identifier rather than a call, and a walker that only follows nested invocations drops - the `extract -> fanin` edge **silently**. The processor resolves an identifier through the local's - initializer, and rejects an argument it cannot attribute rather than emitting an incomplete graph. -- **The getters need no thread-local.** `bind` sets a field on `Tasks`, and the generated code already - constructs the instance, so nothing ambient is required — which matters because the SDK's toolchain - is pinned to Java 11, where `ScopedValue` does not exist. -- **`dag.task(...)` is a factory method, not a constructor.** Java has no `new dag.TaskRef(...)`: - qualified inner-class creation is spelled `dag.new TaskRef(...)`, which reads as compiler trivia - rather than as a Dag definition. -- **`dag.task(...)`, `.config(...)`, `.before(...)`, and `.after(...)` are proposed additions.** +- The processor generates, per Dag, the `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` in declared order, and the return becomes `TaskRef` + (`TaskRef` for `void`). +- **The nested `@Builder.Deps` class is required, not stylistic.** The Dag class holds + `long extract(Client)` and the wiring view holds `TaskRef 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`/`after` are variadic and return 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 two verbs. `Flow.of(...)` + is the only entry point beyond them — it opens a chain from a set, which the verbs cannot. A data + edge never needs any of this — it is implied by passing a `TaskRef`. +- **`Arg` is the data-argument type in a wiring method**; `TaskRef extends Arg`, 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` overload for generics. Nothing is ambient, so the SDK's Java 11 pin (no + `ScopedValue`) does not matter here. +- **`dag.task(...)`, `.config(...)`, `.then(...)`, and `.after(...)` 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. -- **`register` is one overloaded verb**: `register(Class)` for an annotated class, `register(Dag)` - for a built one, and `register(dagId, taskId, Class)` for a task handler - ([ADR-0001](0001-mixed-lang-dag-interface.md)). It is proposed here; today a bundle is built from an + `.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)` for a task + handler ([ADR-0001](0001-mixed-lang-dag-interface.md)). Today a bundle is built from an `Iterable` through `BundleBuilder.getDags()`. +- **Registration holds classes, never instances.** `Dag.addTask(id, Class)` 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` to unwrap. From c7df9240302c7719e3805e73f371383cd417e30b Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Tue, 15 Sep 2026 11:44:28 +0800 Subject: [PATCH 7/8] Remove 'after' from chaining interface --- java-sdk/adr/0002-native-dag-interface.md | 54 +++++++++++------------ 1 file changed, 27 insertions(+), 27 deletions(-) diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index 1b041fee02495..a0838d8088bc1 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -77,7 +77,7 @@ public class EtlPipeline { // extends nothing of ours; your own base class stays // non-TaskFlow (ordering-only) edges: sequence with no data flowing rows.then(audit()); // extract >> audit - notify().after(loaded, audit()); // [load, audit] >> notify + Flow.of(loaded, audit()).then(notify()); // [load, audit] >> notify } } @@ -126,10 +126,13 @@ 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).after(extract); - TaskRef load = dag.task("load", Load.class).after(transform); - TaskRef audit = dag.task("audit", Audit.class).after(extract); // ordering-only edge - dag.task("notify", Notify.class).after(load, audit); // ordering-only edge + 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; } @@ -159,24 +162,23 @@ primitives. ### Non-TaskFlow dependencies A TaskFlow (data) edge comes for free from passing a `TaskRef` into another wiring method. A -dependency where **no data flows** — Python's `a >> b`, and the list forms `a >> [b, c]` and -`[x, y] >> z` — is expressed instead with two variadic verbs that every `TaskRef` carries, `then` -and `after`. Both live on a small `Chain` interface, and a `TaskRef` is a `Chain` of one -(`interface TaskRef extends Arg, Chain`): +dependency where no data flows. Python's `a >> b`, and the list forms `a >> [b, c]` and `[x, y] >> z` are +expressed with `then`, a variadic verb that every `TaskRef` carries. It lives on a small `Chain` interface, +and a `TaskRef` is a `Chain` of one (`interface TaskRef extends Arg, Chain`): ```java +a.then(b); // a >> b a.then(b, c); // a >> [b, c] -z.after(x, y); // z << [x, y] ``` -`then` and `after` are mirror images, and each returns the **new frontier** (the set it just pointed -at), the way `>>`/`<<` evaluate to their right operand, so a chain walks through a fan: +`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 the verbs cannot do is start from a *set*: Java can't overload `>>` the way Python +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 @@ -184,10 +186,7 @@ Flow.of(a, b).then(c); // [a, b] >> c Flow.of(a, b).then(c, d); // [a, b] >> [c, d] ``` -That is the whole non-TaskFlow surface: `then` and `after` on `Chain`, plus `Flow.of`. There is no -`chain`, `fanOut`, or `fanIn` — `a.then(b).then(c)` is a chain, `a.then(b, c)` is a fan-out, and -`z.after(x, y)` is a fan-in — so the shorthands would earn nothing. -Both edge kinds compile to the same `after(...)` in the generated Dag above — the only difference is +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 @@ -196,15 +195,16 @@ that a data edge also binds an argument while an ordering-only edge binds nothin 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).after(extract); -TaskRef load = dag.task("load", Load.class).after(transform); +TaskRef transform = dag.task("transform", Transform.class); +TaskRef load = dag.task("load", Load.class); +TaskRef notify = dag.task("notify", Notify.class); -load.then(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` and `after` are the same variadic edge verbs as the -annotation surface — Java's spelling of Python's `>>` and `<<` — and `.config(key, value)` carries +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. @@ -292,10 +292,10 @@ One bundle carries native Dags and mixed-language task handlers alike `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`/`after` are variadic and return 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 two verbs. `Flow.of(...)` - is the only entry point beyond them — it opens a chain from a set, which the verbs cannot. A data - edge never needs any of this — it is implied by passing a `TaskRef`. +- **`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`. - **`Arg` is the data-argument type in a wiring method**; `TaskRef extends Arg`, 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, @@ -305,7 +305,7 @@ One bundle carries native Dags and mixed-language task handlers alike ([ADR-0001](0001-mixed-lang-dag-interface.md)) — the same conversion-not-cast path, including the `TypeReference` overload for generics. Nothing is ambient, so the SDK's Java 11 pin (no `ScopedValue`) does not matter here. -- **`dag.task(...)`, `.config(...)`, `.then(...)`, and `.after(...)` are proposed additions.** +- **`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 From 2644919751e946999ddce174628a17772e6e0b6b Mon Sep 17 00:00:00 2001 From: Tzu-ping Chung Date: Tue, 15 Sep 2026 13:03:05 +0800 Subject: [PATCH 8/8] Tighten flow API design --- java-sdk/adr/0002-native-dag-interface.md | 38 +++++++++++++++++++---- 1 file changed, 32 insertions(+), 6 deletions(-) diff --git a/java-sdk/adr/0002-native-dag-interface.md b/java-sdk/adr/0002-native-dag-interface.md index a0838d8088bc1..80c2d7101cdbd 100644 --- a/java-sdk/adr/0002-native-dag-interface.md +++ b/java-sdk/adr/0002-native-dag-interface.md @@ -96,7 +96,7 @@ argument lifted to `Arg`, and the return lifted to `TaskRef`: ```java // Generated by the Airflow Java SDK annotation processor. Do not edit. -interface EtlPipelineDeps { +interface EtlPipelineDeps extends Deps { // Deps: the shared base that nests Flow default TaskRef extract() { return Flow.node("extract"); } default TaskRef transform(Arg extracted, Arg threshold) { return Flow.call("transform", extracted, threshold); } default TaskRef load(Arg transformed) { return Flow.call("load", transformed); } @@ -159,12 +159,12 @@ with `lit(...)` — `transform(extract(), lit(0.9))` — and recorded as a baked Wrapping is required because a bare `Integer` cannot implement `Arg`; boxed types only, no primitives. -### Non-TaskFlow dependencies +### Non-data dependencies -A TaskFlow (data) edge comes for free from passing a `TaskRef` into another wiring method. A -dependency where no data flows. Python's `a >> b`, and the list forms `a >> [b, c]` and `[x, y] >> z` are -expressed with `then`, a variadic verb that every `TaskRef` carries. It lives on a small `Chain` interface, -and a `TaskRef` is a `Chain` of one (`interface TaskRef extends Arg, Chain`): +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 extends Arg, Flow`): ```java a.then(b); // a >> b @@ -258,6 +258,21 @@ One bundle carries native Dags and mixed-language task handlers alike 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 `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 @@ -296,6 +311,17 @@ One bundle carries native Dags and mixed-language task handlers alike 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 `Deps` + extends one library interface, `Deps`, that declares the nested interface `Flow`. Member types are + inherited, so a `@Builder.Deps` class implementing `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` is the data-argument type in a wiring method**; `TaskRef extends Arg`, 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,