Skip to content

Java SDK: Rename annotation namespace from Builder to Airflow #74604

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jason810496 merged 1 commit into from
Oct 13, 2026
Merged

Java SDK: Rename annotation namespace from Builder to Airflow #74604

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .agents/skills/airflow-java-sdk/SKILL.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ description: >
coordinator in `task-sdk/src/airflow/sdk/coordinators/java/` — whether they
want to add a feature, write tests, fix a bug, understand the architecture, or
prepare a PR. Trigger on phrases like "Java SDK", "JavaCoordinator",
"java-sdk", "annotation processor", "Builder.Task", "BundleBuilder", or
"java-sdk", "annotation processor", "Airflow.Task", "BundleBuilder", or
anything about running JVM tasks in Airflow.
---

Expand Down Expand Up @@ -87,7 +87,7 @@ the Dag bundle or startup fails. Every JAR in the Dag bundle goes on one classpa
| `java-sdk/sdk/.../execution/Client.kt` | Supervisor wire calls |
| `java-sdk/sdk/.../execution/Comm.kt` | 4-byte-prefix MessagePack framing |
| `java-sdk/sdk/.../Server.kt` | Entry-point; drives the execution loop |
| `java-sdk/processor/.../BuilderProcessor.kt` | Kapt annotation processor |
| `java-sdk/processor/.../AirflowProcessor.kt` | Kapt annotation processor |
| `java-sdk/plugin/.../AirflowSdkPlugin.kt` | Gradle bundle plugin |
| `task-sdk/.../coordinators/java/coordinator.py` | Python side — spawns the JVM |
| `task-sdk/.../schema/schema.json` | Wire protocol definition (both sides) |
Expand Down
70 changes: 35 additions & 35 deletions airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,16 +178,16 @@ Save the task implementations as

package com.mycompany.airflow.sales;

import org.apache.airflow.sdk.Builder;
import org.apache.airflow.sdk.Airflow;

public class SalesPipeline {

@Builder.TaskHandler(dag = "sales_pipeline", task = "extract")
@Airflow.TaskHandler(dag = "sales_pipeline", task = "extract")
public long extract() {
return 3;
}

@Builder.TaskHandler(dag = "sales_pipeline", task = "transform")
@Airflow.TaskHandler(dag = "sales_pipeline", task = "transform")
public long transform(long recordCount) {
return recordCount * 2;
}
Expand Down Expand Up @@ -344,20 +344,20 @@ Annotate a plain Java class and let the SDK generate the boilerplate at compile

* - Annotation
- Purpose
* - ``@Builder.TaskHandler(dag = "...", task = "...")``
* - ``@Airflow.TaskHandler(dag = "...", task = "...")``
- Marks a method as the Java body of a task the Python Dag file declares with ``@task.stub``.
``dag`` must match the ``dag_id`` and ``task`` the stub function name; omitting ``task``
derives it from the method name. There is no class-level annotation on this surface — the
Dag is the Python file's, so the handler names the pair it binds to.
* - ``@Builder.Dag(id = "...")``
* - ``@Airflow.Dag(id = "...")``
- Marks a class as a Dag that Java itself owns. Attributes (``schedule``, ``description``,
``tags``, ``catchup``, …) are Airflow's own Dag settings; only attributes written
explicitly are applied. See :ref:`java-sdk/native-dags`.
* - ``@Builder.Task(id = "...")``
* - ``@Airflow.Task(id = "...")``
- Marks a method as a task of a Java-owned Dag. If ``id`` is omitted the method name is
used. Further attributes (``retries``, ``queue``, ``retryDelay``, …) are Airflow's own
task settings; only attributes written explicitly are applied.
* - ``@Builder.Deps``
* - ``@Airflow.Deps``
- Marks the nested class that declares the task graph in Java, TaskFlow-style. Required for
a Dag that Java owns end to end. See :ref:`java-sdk/native-dags`.
* - ``TaskInput`` / ``@ArgName("...")``
Expand All @@ -378,14 +378,14 @@ registry and resolves data parameters and XCom pushes automatically.

public class MyDag {

@Builder.TaskHandler(dag = "my_dag", task = "fetch")
@Airflow.TaskHandler(dag = "my_dag", task = "fetch")
public String fetch(Client client) throws Exception {
var conn = client.getConnection("my_api");
// implement task logic
return result;
}

@Builder.TaskHandler(dag = "my_dag", task = "process")
@Airflow.TaskHandler(dag = "my_dag", task = "process")
public long process(Client client, String fetched) {
var threshold = (String) client.getVariable("process_threshold");
// implement task logic
Expand Down Expand Up @@ -493,7 +493,7 @@ so renaming one in an IDE never rebinds an input.

.. code-block:: java

@Builder.TaskHandler(dag = "etl", task = "score")
@Airflow.TaskHandler(dag = "etl", task = "score")
public long score(Client client, long rows, double threshold) {
// rows <- the load_rows XCom (position 0)
// threshold <- the literal 0.75 (position 1)
Expand Down Expand Up @@ -552,7 +552,7 @@ and reported rather than handed the wrong value.
public double threshold; // binds threshold
}

@Builder.TaskHandler(dag = "etl", task = "score")
@Airflow.TaskHandler(dag = "etl", task = "score")
public long score(Client client, ScoreInput input) { ... }

Reach for ``@ArgName`` when the argument name is not a legal or usable Java identifier — a Python
Expand Down Expand Up @@ -620,12 +620,12 @@ Edges are checked when the Dag is registered with a ``Bundle``: an upstream that
Dag, or to no Dag, and a cycle anywhere in the graph both fail there rather than at the first task
run.

Wiring the graph with ``@Builder.Deps``
Wiring the graph with ``@Airflow.Deps``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

For a Dag written with annotations, the graph is declared by a nested ``@Builder.Deps`` class. The
For a Dag written with annotations, the graph is declared by a nested ``@Airflow.Deps`` class. The
annotation processor generates a ``<ClassName>Deps`` interface, the *wiring view*, with one method
per ``@Builder.Task`` method: the injected ``Client`` and ``Context`` parameters are dropped, each
per ``@Airflow.Task`` method: the injected ``Client`` and ``Context`` parameters are dropped, each
data parameter becomes an ``Arg<T>``, and the return value becomes a ``TaskRef<T>``. Calling a view
method registers its task, and passing the handle one returned into another feeds the upstream's
output into the downstream's parameter *and* wires the data edge. The call graph is the task graph,
Expand All @@ -636,34 +636,34 @@ generated view, with a no-argument ``depends()`` method:

.. code-block:: java

@Builder.Dag(
@Airflow.Dag(
id = "java_etl",
schedule = "@daily",
description = "Pure-Java Dag built with annotations",
tags = {"example", "java-sdk"})
public class EtlPipeline {

@Builder.Task(id = "extract", retries = 2)
@Airflow.Task(id = "extract", retries = 2)
public long extract() {
return 42L;
}

@Builder.Task(id = "transform")
@Airflow.Task(id = "transform")
public long transform(long extracted, double factor) {
return (long) (extracted * factor);
}

@Builder.Task(id = "load")
@Airflow.Task(id = "load")
public void load(long transformed) {
// implement task logic
}

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

@Builder.Deps
@Airflow.Deps
static class Wiring implements EtlPipelineDeps {
void depends() {
var rows = extract();
Expand All @@ -673,7 +673,7 @@ generated view, with a no-argument ``depends()`` method:
}
}

Every ``@Builder.Task`` method must be called in the wiring class; a task the wiring missed fails at
Every ``@Airflow.Task`` method must be called in the wiring class; a task the wiring missed fails at
Dag-parse time. ``lit(...)`` wires an inline constant where no upstream feeds a parameter. A bare
``double`` cannot be an ``Arg``, so a constant is wrapped. Airflow records what each task is called
with, and that record travels as JSON, so a constant has to be a string, number, boolean, list or
Expand All @@ -686,15 +686,15 @@ handle in a local and reuse that.
the wiring class ``Flow`` is inherited by simple name, so it needs no import and never collides with
``java.util.concurrent.Flow``.

Every ``@Builder.Dag`` class declares a wiring class, because the graph is what the Dag owns. A
Every ``@Airflow.Dag`` class declares a wiring class, because the graph is what the Dag owns. A
class that supplies only task bodies, for a Dag a Python file declares, carries
``@Builder.TaskHandler`` instead and contributes no Dag.
``@Airflow.TaskHandler`` instead and contributes no Dag.

.. note::

A native Java Dag binds its task arguments from its own wiring, and the ``_arg_bindings`` it
serializes are what Airflow records and shows. Runtime bindings (see :ref:`java-sdk/arg-binding`)
are what a ``@Builder.TaskHandler`` class reads, for a task whose Dag a Python file declares.
are what an ``@Airflow.TaskHandler`` class reads, for a task whose Dag a Python file declares.

Task groups
~~~~~~~~~~~
Expand All @@ -711,24 +711,24 @@ the Dag or inside another group, and the group declares its tasks:
staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); // "staging.checks.nulls"
extract.before(staging);

With annotations, a ``@Builder.TaskGroup`` class holds the tasks of one group, and nesting one in
With annotations, an ``@Airflow.TaskGroup`` class holds the tasks of one group, and nesting one in
another nests the groups:

.. code-block:: java

@Builder.TaskGroup // the group "Staging", after the class
@Airflow.TaskGroup // the group "Staging", after the class
static class Staging {
@Builder.Task
@Airflow.Task
public long stage(long rows) { ... } // the task "Staging.stage"

@Builder.TaskGroup(id = "checks")
@Airflow.TaskGroup(id = "checks")
static class Checks {
@Builder.Task
@Airflow.Task
public void nulls(long staged) { ... } // "Staging.checks.nulls"
}
}

@Builder.Deps
@Airflow.Deps
static class Wiring implements EtlPipelineDeps {
void depends() {
var rows = extract();
Expand Down Expand Up @@ -762,7 +762,7 @@ underscores, or dashes, and no task or other group in the Dag can share it.
Configuration attributes
~~~~~~~~~~~~~~~~~~~~~~~~

The ``@Builder.Dag`` and ``@Builder.Task`` configuration attributes, and the keys accepted by
The ``@Airflow.Dag`` and ``@Airflow.Task`` configuration attributes, and the keys accepted by
``DagDef.config`` and ``TaskDef.config``, are Airflow's own Dag and task settings, under the names
Airflow uses. Annotation attributes are ``camelCase`` (``retryDelay``); ``config`` keys are those
names as Airflow writes them (``"retry_delay"``).
Expand All @@ -788,7 +788,7 @@ remember things like an external job ID so a retried task can resume instead of

.. code-block:: java

@Builder.Task(id = "submit")
@Airflow.Task(id = "submit")
public void submit(Client client) throws Exception {
var store = client.getTaskStateStore();
var jobId = (String) store.get("job_id");
Expand Down Expand Up @@ -845,7 +845,7 @@ SDK older than schema ``2026-10-30``, every released one today, fails to parse a
several Java coordinators, an unmapped bundle's JARs fail to parse too.

* Every JAR in the bundle whose manifest sets ``Main-Class`` is parsed. Each Dag its main class
declares, through ``Bundle.register`` of a ``DagDef`` or an ``@Builder.Dag`` class, is stored with
declares, through ``Bundle.register`` of a ``DagDef`` or an ``@Airflow.Dag`` class, is stored with
that JAR as its file. Task handlers for a Python Dag are not Dags. A JAR without ``Main-Class`` is
skipped, but many dependency JARs set one (the PostgreSQL JDBC driver and H2 do, for example), so a
thin bundle should set ``main_class`` or list its dependency JARs in ``.airflowignore``.
Expand All @@ -858,7 +858,7 @@ several Java coordinators, an unmapped bundle's JARs fail to parse too.
``Main-Class`` are parsed. List JARs that should not be parsed in ``.airflowignore``.
* Two JARs in the bundle that set the same ``Main-Class`` fail to parse, because the JVM would load
the classes of only one of them. Keep one in the bundle.
* Set ``queue`` on every task, with ``@Builder.Task(queue = "java-native")`` or
* Set ``queue`` on every task, with ``@Airflow.Task(queue = "java-native")`` or
``TaskDef.config("queue", "java-native")``, so it runs on the coordinator's queue. There is no
Dag-level queue yet.
* A task runs the JAR of its Dag on the ``JavaCoordinator`` that its queue routes to,
Expand Down Expand Up @@ -889,7 +889,7 @@ is the conventional pattern regardless of which logging framework you choose:
private static final System.Logger log =
System.getLogger(SalesPipeline.class.getName());

@Builder.TaskHandler(dag = "etl", task = "extract")
@Airflow.TaskHandler(dag = "etl", task = "extract")
public long extract(Client client) {
log.log(System.Logger.Level.INFO, "Starting extraction");
return recordCount;
Expand Down
28 changes: 14 additions & 14 deletions ...e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,70 +25,70 @@
import org.apache.airflow.sdk.*;

/** A native Java Dag declared with annotations; its tasks use the interface Dag's queue. */
@Builder.Dag(
@Airflow.Dag(
id = "java_native_annotation_e2e",
description = "Native Java Dag of the Airflow E2E tests, declared with annotations",
catchup = false,
tags = {"java-sdk", "native", "e2e"})
public class AnnotationDag {
@Builder.Task(id = "extract", queue = QUEUE)
@Airflow.Task(id = "extract", queue = QUEUE)
public long extract() {
return 42L;
}

@Builder.Task(id = "transform", queue = QUEUE)
@Airflow.Task(id = "transform", queue = QUEUE)
public long transform(long extracted, double factor) {
return (long) (extracted * factor);
}

/** Fails the run unless the value reached it, so a successful run proves the XCom flow. */
@Builder.Task(id = "load", queue = QUEUE)
@Airflow.Task(id = "load", queue = QUEUE)
public void load(long transformed) {
if (transformed != 63L) {
throw new IllegalStateException("load expected 63 from transform, got " + transformed);
}
}

@Builder.Task(id = "report_many", queue = QUEUE)
@Airflow.Task(id = "report_many", queue = QUEUE)
public void reportMany() {}

@Builder.Task(id = "report_few", queue = QUEUE)
@Airflow.Task(id = "report_few", queue = QUEUE)
public void reportFew() {}

/** Picks one of {@link #reportMany} and {@link #reportFew}; the other is skipped. */
@Builder.If(id = "has_rows", queue = QUEUE)
@Airflow.If(id = "has_rows", queue = QUEUE)
public boolean hasRows(long transformed) {
return transformed > 0;
}

@Builder.Task(id = "report_long", queue = QUEUE)
@Airflow.Task(id = "report_long", queue = QUEUE)
public void reportLong() {}

@Builder.Task(id = "report_short", queue = QUEUE)
@Airflow.Task(id = "report_short", queue = QUEUE)
public void reportShort() {}

/** Names the one of {@link #reportLong} and {@link #reportShort} that runs; the other is skipped. */
@Builder.Switch(id = "pick_report", queue = QUEUE)
@Airflow.Switch(id = "pick_report", queue = QUEUE)
public Class<? extends Task> pickReport(long transformed) {
return transformed > 100 ? AnnotationDagBuilder.ReportLong.class : AnnotationDagBuilder.ReportShort.class;
}

// A task that starts a run of another Dag. The method runs when this Dag is
// built, not when the task runs, so it takes no arguments.
@Builder.Task(id = "trigger_downstream", queue = QUEUE)
@Airflow.Task(id = "trigger_downstream", queue = QUEUE)
public TriggerDagRun triggerDownstream() {
return new TriggerDagRun("java_native_target_e2e");
}

// A task group: everything it declares is prefixed with its id, so this is
// the task "checks.audit".
@Builder.TaskGroup(id = "checks")
@Airflow.TaskGroup(id = "checks")
static class Checks {
@Builder.Task(id = "audit", queue = QUEUE)
@Airflow.Task(id = "audit", queue = QUEUE)
public void audit() {}
}

@Builder.Deps
@Airflow.Deps
static class Wiring implements AnnotationDagDeps {
void depends() {
var extracted = extract();
Expand Down
16 changes: 8 additions & 8 deletions java-sdk/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -643,7 +643,7 @@ prek hook regenerate it.
| capability: `task-args` | MUST † | ✓ | 3.4 | |
| capability: `dag-params` | MUST † | ✗ | – | |
| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4 | |
| capability: `branching` | SHOULD † | ✓ | 3.4 | DagDef.If and DagDef.Switch, or @Builder.If and @Builder.Switch |
| capability: `branching` | SHOULD † | ✓ | 3.4 | DagDef.If and DagDef.Switch, or @Airflow.If and @Airflow.Switch |
| capability: `dag-test` | SHOULD † | ✗ | – | |
| capability: `task-group` | MAY † | ✓ | 3.4 | |
| capability: `dynamic-task-mapping` | MAY † | ✗ | – | |
Expand Down Expand Up @@ -746,13 +746,13 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
not the implementation language.
- Keep `sdk/src/main/kotlin/` (the public API surface) free of internal
implementation details; those belong in the `execution/` sub-package.
- The annotation processor (`BuilderProcessor.kt`) uses `kapt`. The `Builder`
class holding the `@Builder.Dag` / `@Builder.Task` annotations is generated
- The annotation processor (`AirflowProcessor.kt`) uses `kapt`. The `Airflow`
class holding the `@Airflow.Dag` / `@Airflow.Task` annotations is generated
from the Dag serialization schema by `:sdk:generateDagDsl` (vendored at
`sdk/schema/dag-schema.json`), `@Builder.Deps` included. The `Arg`/`TaskRef`
`sdk/schema/dag-schema.json`), `@Airflow.Deps` included. The `Arg`/`TaskRef`
and `Deps`/`Flow` graph types are hand-written next to the rest of the public
surface in `sdk/src/main/kotlin/org/apache/airflow/sdk/`. When adding annotation
behaviour, handle it in `BuilderProcessor.kt` and add a golden-output test in
behaviour, handle it in `AirflowProcessor.kt` and add a golden-output test in
`processor/src/test/kotlin/`.
- The Python coordinator subclasses `SubprocessCoordinator`. Do not reach into
the JVM process from Python beyond what `_build_execute_task_command`
Expand All @@ -779,12 +779,12 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
1. Hand-written annotations live in `sdk/src/main/kotlin/org/apache/airflow/sdk/`
next to the runtime types (one package, so user code needs a single
`import org.apache.airflow.sdk.*`); the configuration attributes of
`@Builder.Dag` / `@Builder.Task` come from the Dag serialization schema via
`@Airflow.Dag` / `@Airflow.Task` come from the Dag serialization schema via
`:sdk:generateDagDsl` (adjust its allowlist/exclusion rules in
`sdk/build.gradle.kts` when the exposed field set should change).
2. Handle it in `BuilderProcessor.kt` — generate the appropriate code in the
2. Handle it in `AirflowProcessor.kt` — generate the appropriate code in the
`*Builder` class.
3. Add a test in `BuilderTest.kt` with expected generated output.
3. Add a test in `AirflowProcessorTest.kt` with expected generated output.
4. Update the annotation table in `java.rst`.

**Fixing a framing or protocol bug**: focus on `execution/Comm.kt` and
Expand Down
2 changes: 1 addition & 1 deletion java-sdk/capabilities.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ capabilities:
branching:
supported: true
since: "3.4"
note: "DagDef.If and DagDef.Switch, or @Builder.If and @Builder.Switch"
note: "DagDef.If and DagDef.Switch, or @Airflow.If and @Airflow.Switch"
dag-test:
supported: false
task-group:
Expand Down
Loading
Loading