This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new b3ecf729807 [v3-4-test] Java SDK: Rename annotation namespace from
Builder to Airflow (#74604) (#74826)
b3ecf729807 is described below
commit b3ecf7298077232a47bfaa603a6759187799bce5
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Oct 17 18:49:36 2026 +0530
[v3-4-test] Java SDK: Rename annotation namespace from Builder to Airflow
(#74604) (#74826)
(cherry picked from commit d479721a4a4acda65e7cac1a661dd27f2b92ec9d)
Co-authored-by: Jason(Zhe-You) Liu
<[email protected]>
---
.agents/skills/airflow-java-sdk/SKILL.md | 4 +-
.../language-sdks/java.rst | 70 +--
.../airflow/e2e/nativedag/AnnotationDag.java | 28 +-
java-sdk/README.md | 16 +-
java-sdk/capabilities.yaml | 2 +-
.../apache/airflow/example/AnnotationExample.java | 10 +-
.../apache/airflow/example/XComCastingExample.java | 16 +-
.../example/nativedag/AnnotationExample.java | 30 +-
java-sdk/processor/build.gradle.kts | 2 +-
.../{BuilderProcessor.kt => AirflowProcessor.kt} | 120 ++--
.../services/javax.annotation.processing.Processor | 2 +-
.../{BuilderTest.kt => AirflowProcessorTest.kt} | 682 ++++++++++-----------
java-sdk/sdk/build.gradle.kts | 38 +-
java-sdk/sdk/module.md | 2 +-
.../src/main/kotlin/org/apache/airflow/sdk/Arg.kt | 2 +-
.../main/kotlin/org/apache/airflow/sdk/Bundle.kt | 16 +-
.../main/kotlin/org/apache/airflow/sdk/DagDef.kt | 12 +-
.../src/main/kotlin/org/apache/airflow/sdk/Deps.kt | 4 +-
.../kotlin/org/apache/airflow/sdk/TaskInput.kt | 4 +-
.../kotlin/org/apache/airflow/sdk/TriggerDagRun.kt | 2 +-
.../org/apache/airflow/sdk/internal/ArgValues.kt | 2 +-
.../org/apache/airflow/sdk/internal/Fields.kt | 4 +-
.../org/apache/airflow/sdk/internal/InputTypes.kt | 2 +-
.../kotlin/org/apache/airflow/sdk/internal/Refs.kt | 8 +-
.../org/apache/airflow/sdk/internal/Registrar.kt | 4 +-
.../kotlin/org/apache/airflow/sdk/BundleTest.kt | 12 +-
.../apache/airflow/sdk/internal/ArgValuesTest.kt | 2 +-
.../org/apache/airflow/sdk/internal/RefsTest.kt | 4 +-
.../airflow/k8sexample/MixedLanguageExample.java | 4 +-
task-sdk/docs/lang-sdk-spec.rst | 6 +-
30 files changed, 555 insertions(+), 555 deletions(-)
diff --git a/.agents/skills/airflow-java-sdk/SKILL.md
b/.agents/skills/airflow-java-sdk/SKILL.md
index 9921832eafe..5681c6f7ffd 100644
--- a/.agents/skills/airflow-java-sdk/SKILL.md
+++ b/.agents/skills/airflow-java-sdk/SKILL.md
@@ -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.
---
@@ -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) |
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
index 793eda35412..e8e63811c96 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -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;
}
@@ -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("...")``
@@ -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
@@ -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)
@@ -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
@@ -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,
@@ -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();
@@ -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
@@ -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
~~~~~~~~~~~
@@ -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();
@@ -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"``).
@@ -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");
@@ -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``.
@@ -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,
@@ -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;
diff --git
a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
index e4d8fc71c55..78531c2ccb5 100644
---
a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
+++
b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
@@ -25,70 +25,70 @@ import static
org.apache.airflow.e2e.NativeBundleBuilder.QUEUE;
import org.apache.airflow.sdk.*;
/** A native Java Dag declared with annotations; its tasks use the interface
Dag's queue. */
[email protected](
[email protected](
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();
diff --git a/java-sdk/README.md b/java-sdk/README.md
index fda035f80dc..11f326e3a8e 100644
--- a/java-sdk/README.md
+++ b/java-sdk/README.md
@@ -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 † | ✗ | – | |
@@ -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`
@@ -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
diff --git a/java-sdk/capabilities.yaml b/java-sdk/capabilities.yaml
index 84ee13a9dda..cf9ec1c5fa7 100644
--- a/java-sdk/capabilities.yaml
+++ b/java-sdk/capabilities.yaml
@@ -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:
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
b/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
index c7f08be99a6..2a9059f91ea 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
@@ -34,7 +34,7 @@ import org.apache.airflow.sdk.*;
public class AnnotationExample {
private static final System.Logger log =
System.getLogger(AnnotationExample.class.getName());
- @Builder.TaskHandler(dag = "java_annotation_example", task = "extract")
+ @Airflow.TaskHandler(dag = "java_annotation_example", task = "extract")
public long extractValue(Client client) throws InterruptedException {
log.log(INFO, "Hello from task");
@@ -53,7 +53,7 @@ public class AnnotationExample {
return new Date().getTime();
}
- @Builder.TaskHandler(dag = "java_annotation_example", task = "transform")
+ @Airflow.TaskHandler(dag = "java_annotation_example", task = "transform")
public long transformValue(Client client, long extracted) {
log.log(INFO, "Got XCom from extract: {0}", extracted);
@@ -69,7 +69,7 @@ public class AnnotationExample {
// the task UP_FOR_RETRY -- which only works because the Java SDK now returns
// RetryTask (instead of a terminal FAILED) when ti_context.should_retry is
// set. The retry then runs this task again and it returns normally.
- @Builder.TaskHandler(dag = "java_annotation_example")
+ @Airflow.TaskHandler(dag = "java_annotation_example")
public void load(Context context, long transformed) {
log.log(INFO, "Got XCom from transform: {0}", transformed);
if (context.ti.tryNumber == 1) {
@@ -86,7 +86,7 @@ public class AnnotationExample {
public long transformed;
}
- @Builder.TaskHandler(dag = "java_annotation_example", task = "report")
+ @Airflow.TaskHandler(dag = "java_annotation_example", 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)) {
@@ -95,7 +95,7 @@ public class AnnotationExample {
}
// Verify one supervisor channel can handle client calls across threads.
- @Builder.TaskHandler(dag = "java_annotation_example", task = "concurrent")
+ @Airflow.TaskHandler(dag = "java_annotation_example", task = "concurrent")
public void concurrentClientCalls(Client client) throws Exception {
var pool = Executors.newFixedThreadPool(8);
try {
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
b/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
index 5e15dd778d6..f3608fbd1b3 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
@@ -30,20 +30,20 @@ import org.apache.airflow.sdk.*;
public class XComCastingExample {
private static final System.Logger log =
System.getLogger(XComCastingExample.class.getName());
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_number")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_number")
public int produceNumber() {
log.log(INFO, "Producing int 7");
return 7;
}
// Any primitive numeric type (byte, short, int, long, float, double) and
its boxed form works the same way.
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_long")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_long")
public long widenToLong(long value) {
log.log(INFO, "Got long {0}", value);
return value + 1;
}
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_double")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_double")
public void widenToDouble(double value) {
log.log(INFO, "Got double {0}", value);
if (value != 8.0) {
@@ -51,12 +51,12 @@ public class XComCastingExample {
}
}
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_nothing")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_nothing")
public void produceNothing() {
// Pushes no return_value XCom.
}
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_nullable")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_nullable")
public void consumeNullable(Integer value) {
log.log(INFO, "Got nullable int {0}", value);
if (value != null) {
@@ -64,13 +64,13 @@ public class XComCastingExample {
}
}
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_fraction")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_fraction")
public double produceFraction() {
log.log(INFO, "Producing double 1.5");
return 1.5;
}
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_float")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_float")
public void consumeFloat(float value) {
log.log(INFO, "Got float {0}", value);
if (value != 1.5f) {
@@ -81,7 +81,7 @@ public class XComCastingExample {
// A parameter with type arguments keeps its element type. Reading an element
// is what would fail if it did not: the wire integers decode to Long unless
// the declared List<Double> survives the binding.
- @Builder.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_double_list")
+ @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_double_list")
public void consumeDoubleList(List<Double> values) {
log.log(INFO, "Got list {0}", values);
double total = values.get(0) + values.get(1);
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java
b/java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java
index 0bb55ceacd0..01639dd6f7e 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java
@@ -24,10 +24,10 @@ import static java.lang.System.Logger.Level.INFO;
import org.apache.airflow.sdk.*;
-// A Dag defined entirely in Java, annotation-style. The @Builder.Dag and
-// @Builder.Task attributes carry the configuration and the @Builder.Deps class
+// A Dag defined entirely in Java, annotation-style. The @Airflow.Dag and
+// @Airflow.Task attributes carry the configuration and the @Airflow.Deps class
// declares the graph -- the single place dependencies are defined.
[email protected](
[email protected](
id = "java_native_annotation_example",
description = "Pure-Java Dag authored with annotations",
schedule = "@daily",
@@ -37,48 +37,48 @@ import org.apache.airflow.sdk.*;
public class AnnotationExample {
private static final System.Logger log =
System.getLogger(AnnotationExample.class.getName());
- @Builder.Task(id = "extract", retries = 2)
+ @Airflow.Task(id = "extract", retries = 2)
public long extract() {
log.log(INFO, "Extracting a value");
return 42L;
}
- @Builder.Task(id = "transform")
+ @Airflow.Task(id = "transform")
public long transform(long extracted, double factor) {
log.log(INFO, "Transforming {0} by {1}", extracted, factor);
return (long) (extracted * factor);
}
- @Builder.Task(id = "load")
+ @Airflow.Task(id = "load")
public void load(long transformed) {
log.log(INFO, "Loaded {0}", transformed);
}
- @Builder.Task(id = "load_empty")
+ @Airflow.Task(id = "load_empty")
public void loadEmpty() {
log.log(INFO, "Nothing to load");
}
// A condition: its boolean picks one of the two loads, and the other is
// skipped.
- @Builder.If(id = "has_rows")
+ @Airflow.If(id = "has_rows")
public boolean hasRows(long transformed) {
return transformed > 0;
}
- @Builder.Task(id = "report_long")
+ @Airflow.Task(id = "report_long")
public void reportLong() {
log.log(INFO, "Long report");
}
- @Builder.Task(id = "report_short")
+ @Airflow.Task(id = "report_short")
public void reportShort() {
log.log(INFO, "Short report");
}
// A switch: it names the one case that runs by the class generated for it,
// and every other case is skipped.
- @Builder.Switch(id = "pick_report")
+ @Airflow.Switch(id = "pick_report")
public Class<? extends Task> pickReport(long transformed) {
return transformed > 100
? AnnotationExampleBuilder.ReportLong.class
@@ -87,16 +87,16 @@ public class AnnotationExample {
// 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")
+ @Airflow.Task(id = "trigger_downstream")
public TriggerDagRun triggerDownstream() {
return new TriggerDagRun("java_native_target_example").config("note",
"from the annotation example");
}
// 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")
+ @Airflow.Task(id = "audit")
public void audit() {
log.log(INFO, "Audited the run");
}
@@ -104,7 +104,7 @@ public class AnnotationExample {
// Implements the generated wiring view, so javac type-checks the graph:
// extract() yields a TaskRef<Long>, which is an Arg<Long> for transform.
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements AnnotationExampleDeps {
void depends() {
var extracted = extract();
diff --git a/java-sdk/processor/build.gradle.kts
b/java-sdk/processor/build.gradle.kts
index bddafe1a051..6ec06748e83 100644
--- a/java-sdk/processor/build.gradle.kts
+++ b/java-sdk/processor/build.gradle.kts
@@ -49,7 +49,7 @@ publishing {
name = "Apache Airflow Java SDK — Annotation Processor"
description =
"Annotation processor for the Apache Airflow Java SDK; " +
- "generates *Builder classes from @Builder.Dag-annotated
sources."
+ "generates *Builder classes from @Airflow.Dag-annotated
sources."
}
}
}
diff --git
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
similarity index 92%
rename from
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
rename to
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
index e101c0e8572..666ca2e23e2 100644
---
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
+++
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
@@ -68,24 +68,24 @@ import org.apache.airflow.sdk.internal.builderName as
generatedBuilderName
/**
* @suppress
*
- * Annotation processor for [Builder.Dag].
+ * Annotation processor for [Airflow.Dag].
*
* This is registered as a standard javac processor via
* `META-INF/services/javax.annotation.processing.Processor`; not intended to
be
* instantiated or referenced directly.
*
- * For each class annotated with [Builder.Dag], generates:
+ * For each class annotated with [Airflow.Dag], generates:
*
- * - A `*Builder` class containing one inner class per [Builder.Task]-annotated
+ * - A `*Builder` class containing one inner class per [Airflow.Task]-annotated
* method (implementing [Task]), and a static `build()` that constructs the
- * [DagDef], lowers every explicitly-written `@Builder.Dag` attribute into a
- * `DagDef.config` call, then runs the class's [Builder.Deps] class and
+ * [DagDef], lowers every explicitly-written `@Airflow.Dag` attribute into a
+ * `DagDef.config` call, then runs the class's [Airflow.Deps] class and
* verifies it registered every task.
* - A `*Deps` wiring-view interface whose methods mirror the task methods:
* injectable parameters ([Client],
* [Context]) are dropped, data parameters become [Arg]-typed inputs, and the
* return value becomes a [TaskRef]. Calling one registers the task with its
- * explicitly-written `@Builder.Task` attributes lowered into
`TaskDef.config`
+ * explicitly-written `@Airflow.Task` attributes lowered into
`TaskDef.config`
* calls; passing one call's handle to another wires the dependency edge and
* feeds the upstream's return-value XCom into the downstream's parameter,
* type-checked by javac through the [Arg] / [TaskRef] generics.
@@ -97,47 +97,47 @@ import org.apache.airflow.sdk.internal.builderName as
generatedBuilderName
* forwarded to `client.setXCom`.
*/
@SupportedAnnotationTypes(
- "org.apache.airflow.sdk.Builder.Dag",
- "org.apache.airflow.sdk.Builder.Task",
- "org.apache.airflow.sdk.Builder.If",
- "org.apache.airflow.sdk.Builder.Switch",
- "org.apache.airflow.sdk.Builder.TaskGroup",
- "org.apache.airflow.sdk.Builder.TaskHandler",
- "org.apache.airflow.sdk.Builder.Deps",
+ "org.apache.airflow.sdk.Airflow.Dag",
+ "org.apache.airflow.sdk.Airflow.Task",
+ "org.apache.airflow.sdk.Airflow.If",
+ "org.apache.airflow.sdk.Airflow.Switch",
+ "org.apache.airflow.sdk.Airflow.TaskGroup",
+ "org.apache.airflow.sdk.Airflow.TaskHandler",
+ "org.apache.airflow.sdk.Airflow.Deps",
)
@SupportedSourceVersion(SourceVersion.RELEASE_11)
-class BuilderProcessor : AbstractProcessor() {
+class AirflowProcessor : AbstractProcessor() {
override fun process(
annotations: Set<TypeElement>,
roundEnv: RoundEnvironment,
): Boolean {
if (annotations.isEmpty()) return false
- roundEnv.getElementsAnnotatedWith(Builder.Deps::class.java).forEach { el ->
+ roundEnv.getElementsAnnotatedWith(Airflow.Deps::class.java).forEach { el ->
val owner = el.enclosingElement
- if (owner !is TypeElement ||
owner.getAnnotation(Builder.Dag::class.java) == null) {
+ if (owner !is TypeElement ||
owner.getAnnotation(Airflow.Dag::class.java) == null) {
processingEnv.messager.printMessage(
Diagnostic.Kind.ERROR,
- "@Builder.Deps class '${el.simpleName}' must be nested directly in a
@Builder.Dag class",
+ "@Airflow.Deps class '${el.simpleName}' must be nested directly in
an @Airflow.Dag class",
el,
)
}
}
- roundEnv.getElementsAnnotatedWith(Builder.TaskGroup::class.java).forEach {
el ->
+ roundEnv.getElementsAnnotatedWith(Airflow.TaskGroup::class.java).forEach {
el ->
val owner = el.enclosingElement
val nested =
owner is TypeElement &&
- (owner.getAnnotation(Builder.Dag::class.java) != null ||
owner.getAnnotation(Builder.TaskGroup::class.java) != null)
+ (owner.getAnnotation(Airflow.Dag::class.java) != null ||
owner.getAnnotation(Airflow.TaskGroup::class.java) != null)
if (!nested) {
processingEnv.messager.printMessage(
Diagnostic.Kind.ERROR,
- "@Builder.TaskGroup class '${el.simpleName}' must be nested in a
@Builder.Dag class or in " +
- "another @Builder.TaskGroup class",
+ "@Airflow.TaskGroup class '${el.simpleName}' must be nested in an
@Airflow.Dag class or in " +
+ "another @Airflow.TaskGroup class",
el,
)
}
}
roundEnv
- .getElementsAnnotatedWith(Builder.TaskHandler::class.java)
+ .getElementsAnnotatedWith(Airflow.TaskHandler::class.java)
.mapNotNull { it.enclosingElement as? TypeElement }
.distinct()
.forEach { el ->
@@ -150,7 +150,7 @@ class BuilderProcessor : AbstractProcessor() {
}.onFailure { e -> messager.printMessage(Diagnostic.Kind.ERROR,
e.message ?: "Unknown error", el) }
}
}
-
roundEnv.getElementsAnnotatedWith(Builder.Dag::class.java).filterIsInstance<TypeElement>().forEach
{ el ->
+
roundEnv.getElementsAnnotatedWith(Airflow.Dag::class.java).filterIsInstance<TypeElement>().forEach
{ el ->
with(processingEnv) {
runCatching {
val packageName =
elementUtils.getPackageOf(el).qualifiedName.toString()
@@ -182,7 +182,7 @@ class BuilderProcessor : AbstractProcessor() {
}
/**
- * Generates the registrar for a class of [Builder.TaskHandler] methods: one
+ * Generates the registrar for a class of [Airflow.TaskHandler] methods: one
* [Task] implementation per handler, and a `registerInto` that binds each to
* the Dag and task the annotation names.
*
@@ -191,7 +191,7 @@ class BuilderProcessor : AbstractProcessor() {
*/
private fun buildHandlers(el: TypeElement): TypeSpec {
require(el.enclosingElement !is TypeElement || Modifier.STATIC in
el.modifiers) {
- "Nested class '${el.simpleName}' holding @Builder.TaskHandler methods
must be static"
+ "Nested class '${el.simpleName}' holding @Airflow.TaskHandler methods
must be static"
}
val registrar =
TypeSpec
@@ -210,18 +210,18 @@ class BuilderProcessor : AbstractProcessor() {
val names = mutableSetOf<String>()
for (inner in el.enclosedElements) {
if (inner !is ExecutableElement) continue
- val handler = inner.getAnnotation(Builder.TaskHandler::class.java) ?:
continue
+ val handler = inner.getAnnotation(Airflow.TaskHandler::class.java) ?:
continue
if (inner.isVarArgs) {
throw IllegalArgumentException("Cannot create task from vararg
function ${inner.simpleName}")
}
require(handler.dag.isNotBlank()) {
- "@Builder.TaskHandler on '${inner.simpleName}' must name the Dag the
Python file declares"
+ "@Airflow.TaskHandler on '${inner.simpleName}' must name the Dag the
Python file declares"
}
val decl = TaskDeclaration(inner, handler.task.ifBlank {
inner.simpleName.toString() }, collectDataParams(inner), el)
require(names.add(inner.simpleName.toString())) {
"Class ${el.simpleName} overloads task-handler method
'${inner.simpleName}'; a method's name is " +
"the name of its generated task class, so rename one and keep its
task id with " +
- "@Builder.TaskHandler(task = \"${decl.id}\")"
+ "@Airflow.TaskHandler(task = \"${decl.id}\")"
}
registrar.addType(buildTask(decl))
registerInto.addStatement(
@@ -234,7 +234,7 @@ class BuilderProcessor : AbstractProcessor() {
return registrar.addMethod(registerInto.build()).build()
}
- private fun dagAnnotation(el: TypeElement): Builder.Dag =
el.getAnnotation(Builder.Dag::class.java)!!
+ private fun dagAnnotation(el: TypeElement): Airflow.Dag =
el.getAnnotation(Airflow.Dag::class.java)!!
private fun buildBuilder(
el: TypeElement,
@@ -287,7 +287,7 @@ class BuilderProcessor : AbstractProcessor() {
* injected arguments stripped, each data argument lifted to [Arg] and the
* return lifted to [TaskRef].
*
- * It is an interface so the `@Builder.Deps` class can *implement* it and
+ * It is an interface so the `@Airflow.Deps` class can *implement* it and
* keep its own `extends` free, and so the Dag class's real task methods --
* which differ only in their injected arguments -- do not clash with it.
*/
@@ -405,7 +405,7 @@ class BuilderProcessor : AbstractProcessor() {
/**
* Emits `new TaskDef(id, <classRef>.class)` with the explicitly-written
- * `@Builder.Task` attributes lowered into chained `.config` calls.
+ * `@Airflow.Task` attributes lowered into chained `.config` calls.
*/
private fun taskDefCode(
decl: TaskDeclaration,
@@ -465,12 +465,12 @@ class BuilderProcessor : AbstractProcessor() {
checkDeciderReturn(kind, inner)
if (kind == TaskKind.TRIGGER) {
require(inner.parameters.isEmpty()) {
- "@Builder.Task method '${inner.simpleName}' returns a TriggerDagRun,
so it runs when the Dag " +
+ "@Airflow.Task method '${inner.simpleName}' returns a TriggerDagRun,
so it runs when the Dag " +
"is built rather than when the task runs; it takes no parameters"
}
val checked = inner.thrownTypes.filterNot { isUnchecked(it) }
require(checked.isEmpty()) {
- "@Builder.Task method '${inner.simpleName}' returns a TriggerDagRun,
so it runs when the Dag " +
+ "@Airflow.Task method '${inner.simpleName}' returns a TriggerDagRun,
so it runs when the Dag " +
"is built, where a checked exception cannot be thrown; it must not
throw: ${checked.joinToString()}"
}
}
@@ -486,7 +486,7 @@ class BuilderProcessor : AbstractProcessor() {
val groups = mutableListOf<GroupDeclaration>()
for (inner in el.enclosedElements.filterIsInstance<TypeElement>()) {
- val ann = inner.getAnnotation(Builder.TaskGroup::class.java) ?: continue
+ val ann = inner.getAnnotation(Airflow.TaskGroup::class.java) ?: continue
val localId = checkGroupClass(inner, ann)
require(groups.none { it.id == localId }) {
"Class ${el.simpleName} declares more than one task group '$localId'"
@@ -500,21 +500,21 @@ class BuilderProcessor : AbstractProcessor() {
/** Checks that `new <group class>()` compiles and names a valid group, and
returns its local ID. */
private fun checkGroupClass(
el: TypeElement,
- ann: Builder.TaskGroup,
+ ann: Airflow.TaskGroup,
): String {
val name = el.simpleName
require(el.kind == ElementKind.CLASS && Modifier.ABSTRACT !in
el.modifiers) {
- "@Builder.TaskGroup '$name' must be a concrete class"
+ "@Airflow.TaskGroup '$name' must be a concrete class"
}
require(Modifier.STATIC in el.modifiers && Modifier.PRIVATE !in
el.modifiers) {
- "@Builder.TaskGroup class '$name' must be static and non-private"
+ "@Airflow.TaskGroup class '$name' must be static and non-private"
}
require(
el.enclosedElements
.filterIsInstance<ExecutableElement>()
.any { it.kind == ElementKind.CONSTRUCTOR && it.parameters.isEmpty()
&& Modifier.PRIVATE !in it.modifiers },
) {
- "@Builder.TaskGroup class '$name' needs a non-private no-argument
constructor"
+ "@Airflow.TaskGroup class '$name' needs a non-private no-argument
constructor"
}
val id = ann.id.ifBlank { name.toString() }
require(GROUP_ID.matches(id)) {
@@ -524,7 +524,7 @@ class BuilderProcessor : AbstractProcessor() {
}
/**
- * Finds and validates the class's `@Builder.Deps` wiring class, which
+ * Finds and validates the class's `@Airflow.Deps` wiring class, which
* declares the Dag's task graph and is what makes it a Dag Java owns.
*
* The generated builder runs `new Wiring()::depends`, so everything that
@@ -537,34 +537,34 @@ class BuilderProcessor : AbstractProcessor() {
val classes =
el.enclosedElements
.filterIsInstance<TypeElement>()
- .filter { it.getAnnotation(Builder.Deps::class.java) != null }
+ .filter { it.getAnnotation(Airflow.Deps::class.java) != null }
require(classes.isNotEmpty()) {
- "Dag class ${el.simpleName} must declare a @Builder.Deps class
implementing ${view.simpleName()} " +
+ "Dag class ${el.simpleName} must declare an @Airflow.Deps class
implementing ${view.simpleName()} " +
"to declare its task graph; a class of task bodies for a Dag the
Python file owns carries " +
- "@Builder.TaskHandler instead"
+ "@Airflow.TaskHandler instead"
}
val deps =
classes.singleOrNull()
?: throw IllegalArgumentException(
- "Dag class ${el.simpleName} declares more than one @Builder.Deps
class: " +
+ "Dag class ${el.simpleName} declares more than one @Airflow.Deps
class: " +
classes.joinToString { it.simpleName.toString() },
)
val name = deps.simpleName
require(deps.kind == ElementKind.CLASS && Modifier.ABSTRACT !in
deps.modifiers) {
- "@Builder.Deps '$name' must be a concrete class"
+ "@Airflow.Deps '$name' must be a concrete class"
}
require(Modifier.STATIC in deps.modifiers && Modifier.PRIVATE !in
deps.modifiers) {
- "@Builder.Deps class '$name' must be static and non-private"
+ "@Airflow.Deps class '$name' must be static and non-private"
}
require(deps.interfaces.any { it.isView(view) }) {
- "@Builder.Deps class '$name' must implement ${view.simpleName()}, the
wiring view of ${el.simpleName}"
+ "@Airflow.Deps class '$name' must implement ${view.simpleName()}, the
wiring view of ${el.simpleName}"
}
require(
deps.enclosedElements
.filterIsInstance<ExecutableElement>()
.any { it.kind == ElementKind.CONSTRUCTOR && it.parameters.isEmpty()
&& Modifier.PRIVATE !in it.modifiers },
) {
- "@Builder.Deps class '$name' needs a non-private no-argument constructor"
+ "@Airflow.Deps class '$name' needs a non-private no-argument constructor"
}
val depends =
processingEnv.elementUtils
@@ -572,11 +572,11 @@ class BuilderProcessor : AbstractProcessor() {
.filterIsInstance<ExecutableElement>()
.firstOrNull { it.isNoArgDepends() }
?: throw IllegalArgumentException(
- "@Builder.Deps class '$name' must have a non-private, no-argument
depends() method",
+ "@Airflow.Deps class '$name' must have a non-private, no-argument
depends() method",
)
val checked = depends.thrownTypes.filterNot { isUnchecked(it) }
require(checked.isEmpty()) {
- "depends() of @Builder.Deps class '$name' must not throw checked
exceptions: ${checked.joinToString()}"
+ "depends() of @Airflow.Deps class '$name' must not throw checked
exceptions: ${checked.joinToString()}"
}
return deps
}
@@ -602,7 +602,7 @@ class BuilderProcessor : AbstractProcessor() {
byClassName.put(decl.className, decl)?.let { first ->
throw IllegalArgumentException(
"Task methods '${first.id}' and '${decl.id}' both generate the task
class " +
- "'${decl.className}'; rename one of them or an enclosing
@Builder.TaskGroup class",
+ "'${decl.className}'; rename one of them or an enclosing
@Airflow.TaskGroup class",
)
}
}
@@ -630,7 +630,7 @@ class BuilderProcessor : AbstractProcessor() {
scope.groups.forEach { group ->
require(group.accessor !in reserved) {
"Task group class '${group.element.simpleName}' clashes with a member
of the wiring view; " +
- "rename the class and keep the group id with @Builder.TaskGroup(id =
\"${group.id}\")"
+ "rename the class and keep the group id with @Airflow.TaskGroup(id =
\"${group.id}\")"
}
require(scope.tasks.none {
it.method.simpleName.contentEquals(group.accessor) }) {
"Task group class '${group.element.simpleName}' and task method
'${group.accessor}' would both " +
@@ -749,12 +749,12 @@ class BuilderProcessor : AbstractProcessor() {
TaskKind.TASK, TaskKind.TRIGGER -> return
TaskKind.CONDITION ->
require(returns.kind == TypeKind.BOOLEAN || with(processingEnv) {
isType(returns, BOXED_BOOLEAN_TYPE) }) {
- "@Builder.If method '${method.simpleName}' returns $returns, but a
condition returns boolean: " +
+ "@Airflow.If method '${method.simpleName}' returns $returns, but a
condition returns boolean: " +
"true runs the task named by Then, false the one named by Else"
}
TaskKind.SWITCH ->
require(returnsTaskClass(returns)) {
- "@Builder.Switch method '${method.simpleName}' returns $returns, but
a switch returns the " +
+ "@Airflow.Switch method '${method.simpleName}' returns $returns, but
a switch returns the " +
"class of the task it chose, as Class<? extends Task>"
}
}
@@ -975,7 +975,7 @@ private class Scope(
fun allGroups(): List<GroupDeclaration> = groups.flatMap { listOf(it) +
it.scope.allGroups() }
}
-/** One `@Builder.TaskGroup` class, and what it declares. */
+/** One `@Airflow.TaskGroup` class, and what it declares. */
private class GroupDeclaration(
val element: TypeElement,
val id: String,
@@ -987,7 +987,7 @@ private class GroupDeclaration(
}
/**
- * One [Builder.Task]-annotated method with its resolved id and data
parameters.
+ * One [Airflow.Task]-annotated method with its resolved id and data
parameters.
* [owner] is the class that declares it, which the generated body
instantiates,
* and [classPath] the task-group classes enclosing it.
*/
@@ -1043,7 +1043,7 @@ private val DEPS_TYPE = ClassName.get(Deps::class.java)
private val GROUP_TYPE = DEPS_TYPE.nestedClass("TaskGroup")
private val LIST_TYPE = ClassName.get(List::class.java)
-private const val DAG_ANNOTATION = "org.apache.airflow.sdk.Builder.Dag"
+private const val DAG_ANNOTATION = "org.apache.airflow.sdk.Airflow.Dag"
/**
* What a declared task is, and the annotation that declares it. A decider
@@ -1058,10 +1058,10 @@ private enum class TaskKind(
val taskInterface: ClassName,
val refType: ClassName?,
) {
- TASK("org.apache.airflow.sdk.Builder.Task", "@Builder.Task", "execute",
TASK_TYPE, null),
- TRIGGER("org.apache.airflow.sdk.Builder.Task", "@Builder.Task", "execute",
TASK_TYPE, null),
- CONDITION("org.apache.airflow.sdk.Builder.If", "@Builder.If", "decide",
CONDITION_TASK_TYPE, CONDITION_REF_TYPE),
- SWITCH("org.apache.airflow.sdk.Builder.Switch", "@Builder.Switch", "choose",
SWITCH_TASK_TYPE, SWITCH_REF_TYPE),
+ TASK("org.apache.airflow.sdk.Airflow.Task", "@Airflow.Task", "execute",
TASK_TYPE, null),
+ TRIGGER("org.apache.airflow.sdk.Airflow.Task", "@Airflow.Task", "execute",
TASK_TYPE, null),
+ CONDITION("org.apache.airflow.sdk.Airflow.If", "@Airflow.If", "decide",
CONDITION_TASK_TYPE, CONDITION_REF_TYPE),
+ SWITCH("org.apache.airflow.sdk.Airflow.Switch", "@Airflow.Switch", "choose",
SWITCH_TASK_TYPE, SWITCH_REF_TYPE),
;
companion object {
diff --git
a/java-sdk/processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor
b/java-sdk/processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor
index cc7c3d49ef7..c383ac1ff8e 100644
---
a/java-sdk/processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor
+++
b/java-sdk/processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor
@@ -15,4 +15,4 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-org.apache.airflow.sdk.BuilderProcessor
+org.apache.airflow.sdk.AirflowProcessor
diff --git
a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
similarity index 85%
rename from
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
rename to
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
index 5d8c89a058b..368b9a58532 100644
--- a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
+++
b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
@@ -30,7 +30,7 @@ import java.nio.file.Files
import javax.tools.JavaFileObject
private fun compile(source: String) =
- Compiler.javac().withProcessors(BuilderProcessor()).compile(
+ Compiler.javac().withProcessors(AirflowProcessor()).compile(
JavaFileObjects.forSourceString("org.apache.airflow.example.TestExample",
source),
)
@@ -41,7 +41,7 @@ private fun JavaFileObjectSubject.hasSourceEquivalentTo(
JavaFileObjects.forSourceString(qual, source),
)
-class BuilderTest {
+class AirflowProcessorTest {
@Test
@DisplayName("generate builder and task-reference twins for dag class")
fun generateBuilderAndRefForDagClass() {
@@ -50,26 +50,26 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Client;
import org.apache.airflow.sdk.Context;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void t1() {}
- @Builder.Task
+ @Airflow.Task
public int t2(Client client) {
return 7;
}
- @Builder.Task
+ @Airflow.Task
public void t3(Context ctx, int value) {
System.out.println(String.format("%s %s", ctx.ti, value));
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
t1();
@@ -178,15 +178,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Client;
import org.apache.airflow.sdk.Context;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void t(long first, Client client, String second, Context ctx,
Integer third) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(lit(1L), lit("second"), lit(3)); }
}
@@ -246,13 +246,13 @@ class BuilderTest {
package org.apache.airflow.example;
import java.util.List;
import java.util.Map;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void t(boolean flag, float fraction, Double boxed,
List<String> tags, Map raw) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(lit(true), lit(1f), lit(2.0),
lit(List.of("a")), lit(Map.of())); }
}
@@ -316,25 +316,25 @@ class BuilderTest {
"""
package org.apache.airflow.example;
import java.util.List;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public String ps() { return "x"; }
- @Builder.Task
+ @Airflow.Task
public void pv() {}
- @Builder.Task
+ @Airflow.Task
public List<String> pl() { return null; }
- @Builder.Task
+ @Airflow.Task
public long pn() { return 1L; }
- @Builder.Task
+ @Airflow.Task
public void t(String text, Object anything, List<String> items, Long
boxed) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
t(ps(), pv(), pl(), pn());
@@ -402,14 +402,14 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag(id = "cfg", schedule = "@daily", tags = {"a", "b"},
catchup = true,
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(id = "cfg", schedule = "@daily", tags = {"a", "b"},
catchup = true,
startDate = "2026-01-01T00:00:00Z")
public class TestExample {
- @Builder.Task(retries = 2, queue = "q", retryDelay = "PT5M",
retryExponentialBackoff = 1.5)
+ @Airflow.Task(retries = 2, queue = "q", retryDelay = "PT5M",
retryExponentialBackoff = 1.5)
public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
t1();
@@ -494,13 +494,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void t(String args, int other) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(lit("a"), lit(1)); }
}
@@ -556,21 +556,21 @@ class BuilderTest {
"""
package org.apache.airflow.example;
import java.util.List;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
public double threshold;
}
- @Builder.Task
+ @Airflow.Task
public void flat(String client, int context) {}
- @Builder.Task
+ @Airflow.Task
public void named(ScoreInput context) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { flat(lit("a"), lit(1)); named(lit(null)); }
}
@@ -635,9 +635,9 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class BaseInput {
private String secret;
@@ -647,7 +647,7 @@ class BuilderTest {
public double threshold;
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput input) {}
}
""",
@@ -665,9 +665,9 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class BaseInput {
public String regionCode;
@@ -677,7 +677,7 @@ class BuilderTest {
public String region_code;
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput input) {}
}
""",
@@ -696,15 +696,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
public double threshold;
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput input, int extra) {}
}
""",
@@ -722,15 +722,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
public double threshold;
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput first, ScoreInput second) {}
}
""",
@@ -748,15 +748,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
double threshold;
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput input) {}
}
""",
@@ -774,9 +774,9 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
public double threshold;
@@ -784,7 +784,7 @@ class BuilderTest {
public ScoreInput(double threshold) { this.threshold = threshold; }
}
- @Builder.Task
+ @Airflow.Task
public void t(ScoreInput input) {}
}
""",
@@ -802,10 +802,10 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag(id = "foo")
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(id = "foo")
public class TestExample {
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -839,10 +839,10 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag(to = "Foo")
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(to = "Foo")
public class TestExample {
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -876,12 +876,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "foo") public void t1() {}
+ @Airflow.Task(id = "foo") public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
t1();
@@ -959,16 +959,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public String ps() { return "x"; }
- @Builder.Task
+ @Airflow.Task
public void t(int v) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(ps()); }
}
@@ -986,16 +986,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public double ratio() { return 2.7; }
- @Builder.Task
+ @Airflow.Task
public void load(long rows) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { load(ratio()); }
}
@@ -1013,18 +1013,18 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
""",
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "Dag class TestExample must declare a @Builder.Deps class implementing
TestExampleDeps " +
+ "Dag class TestExample must declare an @Airflow.Deps class implementing
TestExampleDeps " +
"to declare its task graph; a class of task bodies for a Dag the
Python file owns carries " +
- "@Builder.TaskHandler instead",
+ "@Airflow.TaskHandler instead",
)
}
@@ -1035,17 +1035,17 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class One implements TestExampleDeps {
void depends() { t1(); }
}
- @Builder.Deps
+ @Airflow.Deps
static class Two implements TestExampleDeps {
void depends() {}
}
@@ -1054,7 +1054,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "Dag class TestExample declares more than one @Builder.Deps class: One,
Two",
+ "Dag class TestExample declares more than one @Airflow.Deps class: One,
Two",
)
}
@@ -1065,12 +1065,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
class Wiring implements TestExampleDeps {
void depends() { t1(); }
}
@@ -1079,7 +1079,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Deps class 'Wiring' must be static and non-private",
+ "@Airflow.Deps class 'Wiring' must be static and non-private",
)
}
@@ -1090,12 +1090,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void wireItUp() { t1(); }
}
@@ -1104,7 +1104,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Deps class 'Wiring' must have a non-private, no-argument
depends() method",
+ "@Airflow.Deps class 'Wiring' must have a non-private, no-argument
depends() method",
)
}
@@ -1115,22 +1115,22 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements OtherDeps {
void depends() { t1(); }
}
}
- @Builder.Dag
+ @Airflow.Dag
class Other {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements OtherDeps {
void depends() { t1(); }
}
@@ -1139,7 +1139,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Deps class 'Wiring' must implement TestExampleDeps, the wiring
view of TestExample",
+ "@Airflow.Deps class 'Wiring' must implement TestExampleDeps, the wiring
view of TestExample",
)
}
@@ -1150,12 +1150,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
abstract static class Wiring implements TestExampleDeps {
void depends() { t1(); }
}
@@ -1163,7 +1163,7 @@ class BuilderTest {
""",
)
assertThat(compilation).failed()
- assertThat(compilation).hadErrorContaining("@Builder.Deps 'Wiring' must be
a concrete class")
+ assertThat(compilation).hadErrorContaining("@Airflow.Deps 'Wiring' must be
a concrete class")
}
@Test
@@ -1173,12 +1173,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
Wiring(int unused) {}
void depends() { t1(); }
@@ -1188,7 +1188,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Deps class 'Wiring' needs a non-private no-argument
constructor",
+ "@Airflow.Deps class 'Wiring' needs a non-private no-argument
constructor",
)
}
@@ -1199,12 +1199,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() throws java.io.IOException { t1(); }
}
@@ -1213,7 +1213,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "depends() of @Builder.Deps class 'Wiring' must not throw checked
exceptions: java.io.IOException",
+ "depends() of @Airflow.Deps class 'Wiring' must not throw checked
exceptions: java.io.IOException",
)
}
@@ -1224,16 +1224,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
static class Base implements TestExampleDeps {
void depends() { t1(); }
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring extends Base implements TestExampleDeps {}
}
""",
@@ -1248,9 +1248,9 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
- @Builder.Deps
+ @Airflow.Deps
static class Wiring {
void depends() {}
}
@@ -1259,7 +1259,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Deps class 'Wiring' must be nested directly in a @Builder.Dag
class",
+ "@Airflow.Deps class 'Wiring' must be nested directly in an @Airflow.Dag
class",
)
}
@@ -1270,12 +1270,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "wire") public void depends() {}
+ @Airflow.Task(id = "wire") public void depends() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1285,7 +1285,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
"Task method 'depends' clashes with a member of the wiring view; rename
the method and keep " +
- "the task id with @Builder.Task(id = \"wire\")",
+ "the task id with @Airflow.Task(id = \"wire\")",
)
}
@@ -1296,22 +1296,22 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void extract() {}
+ @Airflow.Task public void extract() {}
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void stage() {}
+ @Airflow.Task public void stage() {}
- @Builder.TaskGroup(id = "checks")
+ @Airflow.TaskGroup(id = "checks")
static class Checks {
- @Builder.Task public void nulls() {}
+ @Airflow.Task public void nulls() {}
}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
extract().before(staging());
@@ -1351,20 +1351,20 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class First {
- @Builder.Task public void run() {}
+ @Airflow.Task public void run() {}
}
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Second {
- @Builder.Task public void run() {}
+ @Airflow.Task public void run() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { first().run().before(second().run()); }
}
@@ -1386,12 +1386,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup(id = "staging.checks")
+ @Airflow.TaskGroup(id = "staging.checks")
static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
}
""",
@@ -1409,19 +1409,19 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
}
""",
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.TaskGroup class 'Staging' must be static and non-private",
+ "@Airflow.TaskGroup class 'Staging' must be static and non-private",
)
}
@@ -1432,17 +1432,17 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void staging() {}
+ @Airflow.Task public void staging() {}
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1463,18 +1463,18 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
abstract static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
}
""",
)
assertThat(compilation).failed()
- assertThat(compilation).hadErrorContaining("@Builder.TaskGroup 'Staging'
must be a concrete class")
+ assertThat(compilation).hadErrorContaining("@Airflow.TaskGroup 'Staging'
must be a concrete class")
}
@Test
@@ -1484,21 +1484,21 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
Staging(String name) {}
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
}
""",
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.TaskGroup class 'Staging' needs a non-private no-argument
constructor",
+ "@Airflow.TaskGroup class 'Staging' needs a non-private no-argument
constructor",
)
}
@@ -1509,15 +1509,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void nodes() {}
+ @Airflow.Task public void nodes() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1527,7 +1527,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
"Task method 'nodes' clashes with a member of the wiring view; rename
the method and keep the " +
- "task id with @Builder.Task(id = \"nodes\")",
+ "task id with @Airflow.Task(id = \"nodes\")",
)
}
@@ -1538,15 +1538,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void before() {}
+ @Airflow.Task public void before() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1556,7 +1556,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
"Task method 'before' clashes with a member of the wiring view; rename
the method and keep the " +
- "task id with @Builder.Task(id = \"before\")",
+ "task id with @Airflow.Task(id = \"before\")",
)
}
@@ -1567,20 +1567,20 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup(id = "checks")
+ @Airflow.TaskGroup(id = "checks")
static class Alpha {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
- @Builder.TaskGroup(id = "checks")
+ @Airflow.TaskGroup(id = "checks")
static class Beta {
- @Builder.Task public void t2() {}
+ @Airflow.Task public void t2() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1598,12 +1598,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task public void nodes() {}
+ @Airflow.Task public void nodes() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
public void depends() { nodes(); }
}
@@ -1620,20 +1620,20 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
- @Builder.TaskGroup(id = "lower")
+ @Airflow.TaskGroup(id = "lower")
static class staging {
- @Builder.Task public void t2() {}
+ @Airflow.Task public void t2() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1653,12 +1653,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "staging.stage") public void stage() {}
+ @Airflow.Task(id = "staging.stage") public void stage() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
public void depends() { stage(); }
}
@@ -1675,17 +1675,17 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "Staging") public void staged() {}
+ @Airflow.Task(id = "Staging") public void staged() {}
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1705,23 +1705,23 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.TaskGroup(id = "flat")
+ @Airflow.TaskGroup(id = "flat")
static class A_B {
- @Builder.Task public void c() {}
+ @Airflow.Task public void c() {}
}
- @Builder.TaskGroup(id = "outer")
+ @Airflow.TaskGroup(id = "outer")
static class A {
- @Builder.TaskGroup(id = "inner")
+ @Airflow.TaskGroup(id = "inner")
static class B {
- @Builder.Task public void c() {}
+ @Airflow.Task public void c() {}
}
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1730,7 +1730,7 @@ class BuilderTest {
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "both generate the task class 'A_B_C'; rename one of them or an
enclosing @Builder.TaskGroup class",
+ "both generate the task class 'A_B_C'; rename one of them or an
enclosing @Airflow.TaskGroup class",
)
}
@@ -1741,19 +1741,19 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
- @Builder.TaskGroup
+ @Airflow.TaskGroup
static class Staging {
- @Builder.Task public void t1() {}
+ @Airflow.Task public void t1() {}
}
}
""",
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.TaskGroup class 'Staging' must be nested in a @Builder.Dag
class or in another " +
- "@Builder.TaskGroup class",
+ "@Airflow.TaskGroup class 'Staging' must be nested in an @Airflow.Dag
class or in another " +
+ "@Airflow.TaskGroup class",
)
}
@@ -1764,9 +1764,9 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
- public class TestExample { @Builder.Task(id = "foo") public void
t1(String... client) {} }
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
+ public class TestExample { @Airflow.Task(id = "foo") public void
t1(String... client) {} }
""",
)
assertThat(compilation).failed()
@@ -1782,13 +1782,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "x")
+ @Airflow.Task(id = "x")
public void t1() {}
- @Builder.Task(id = "x")
+ @Airflow.Task(id = "x")
public void t2() {}
}
""",
@@ -1804,13 +1804,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "a") public void extract() {}
- @Builder.Task(id = "b") public void extract(String text) {}
+ @Airflow.Task(id = "a") public void extract() {}
+ @Airflow.Task(id = "b") public void extract(String text) {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -1821,7 +1821,7 @@ class BuilderTest {
assertThat(compilation).hadErrorContaining(
"Class TestExample overloads task method 'extract'; a method's name is
the name of its " +
"generated task class and of its wiring-view method, so rename one and
keep its task id " +
- "with @Builder.Task(id = \"b\")",
+ "with @Airflow.Task(id = \"b\")",
)
}
@@ -1832,10 +1832,10 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
- @Builder.TaskHandler(dag = "etl", task = "a") public void score() {}
- @Builder.TaskHandler(dag = "etl", task = "b") public void
score(String text) {}
+ @Airflow.TaskHandler(dag = "etl", task = "a") public void score() {}
+ @Airflow.TaskHandler(dag = "etl", task = "b") public void
score(String text) {}
}
""",
)
@@ -1843,7 +1843,7 @@ class BuilderTest {
assertThat(compilation).hadErrorContaining(
"Class TestExample overloads task-handler method 'score'; a method's
name is the name of its " +
"generated task class, so rename one and keep its task id with " +
- "@Builder.TaskHandler(task = \"b\")",
+ "@Airflow.TaskHandler(task = \"b\")",
)
}
@@ -1854,12 +1854,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(retryDelay = "5 minutes") public void t() {}
+ @Airflow.Task(retryDelay = "5 minutes") public void t() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(); }
}
@@ -1879,12 +1879,12 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag(tags = {"say \"hi\"", "back\\slash"})
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(tags = {"say \"hi\"", "back\\slash"})
public class TestExample {
- @Builder.Task public void t() {}
+ @Airflow.Task public void t() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { t(); }
}
@@ -1906,13 +1906,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task(id = "say \"hi\"") public void a() {}
- @Builder.Task(id = "back\\slash") public void b() {}
+ @Airflow.Task(id = "say \"hi\"") public void a() {}
+ @Airflow.Task(id = "back\\slash") public void b() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
a();
@@ -1939,10 +1939,10 @@ class BuilderTest {
package org.apache.airflow.example;
import java.util.List;
import org.apache.airflow.sdk.ArgName;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Client;
import org.apache.airflow.sdk.TaskInput;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
public static class ScoreInput implements TaskInput {
@ArgName("region_code") public String region;
@@ -1950,10 +1950,10 @@ class BuilderTest {
public List<String> tags;
}
- @Builder.Task
+ @Airflow.Task
public double score(Client client, ScoreInput input) { return
input.threshold; }
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() { score(lit(null)); }
}
@@ -2005,13 +2005,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Client;
public class TestExample {
- @Builder.TaskHandler(dag = "etl", task = "score")
+ @Airflow.TaskHandler(dag = "etl", task = "score")
public long score(Client client, long rows) { return rows; }
- @Builder.TaskHandler(dag = "etl")
+ @Airflow.TaskHandler(dag = "etl")
public void audit() {}
}
""",
@@ -2070,16 +2070,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
- @Builder.TaskHandler(dag = "")
+ @Airflow.TaskHandler(dag = "")
public void t() {}
}
""",
)
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.TaskHandler on 't' must name the Dag the Python file declares",
+ "@Airflow.TaskHandler on 't' must name the Dag the Python file declares",
)
}
@@ -2090,10 +2090,10 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
public static class Inner {
- @Builder.TaskHandler(dag = "etl", task = "score")
+ @Airflow.TaskHandler(dag = "etl", task = "score")
public long score(long rows) { return rows; }
}
}
@@ -2145,10 +2145,10 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
public class TestExample {
public class Inner {
- @Builder.TaskHandler(dag = "etl")
+ @Airflow.TaskHandler(dag = "etl")
public void t() {}
}
}
@@ -2157,7 +2157,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "Nested class 'Inner' holding @Builder.TaskHandler methods must be
static",
+ "Nested class 'Inner' holding @Airflow.TaskHandler methods must be
static",
)
}
@@ -2169,19 +2169,19 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.If(id = "has_rows")
+ @Airflow.If(id = "has_rows")
public boolean hasRows() {
return true;
}
- @Builder.Task
+ @Airflow.Task
public void load() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
hasRows().Then(load());
@@ -2267,25 +2267,25 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void extract() {}
- @Builder.If(id = "has_rows")
+ @Airflow.If(id = "has_rows")
public boolean hasRows() {
return true;
}
- @Builder.Task
+ @Airflow.Task
public void load() {}
- @Builder.Task
+ @Airflow.Task
public void reportEmpty() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
extract().before(hasRows());
@@ -2308,23 +2308,23 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Task;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.Switch(id = "pick_path")
+ @Airflow.Switch(id = "pick_path")
public Class<? extends Task> pickPath() {
return TestExampleBuilder.HandleLong.class;
}
- @Builder.Task
+ @Airflow.Task
public void handleLong() {}
- @Builder.Task
+ @Airflow.Task
public void handleShort() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
pickPath().Case(handleLong());
@@ -2346,23 +2346,23 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.Task;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.Switch(id = "pick_path")
+ @Airflow.Switch(id = "pick_path")
public Class<? extends Task> pickPath() {
return TestExampleBuilder.HandleLong.class;
}
- @Builder.Task
+ @Airflow.Task
public void handleLong() {}
- @Builder.Task
+ @Airflow.Task
public void handleShort() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
pickPath().Case(handleLong()).Case(handleShort());
@@ -2460,17 +2460,17 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TriggerDagRun;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.Task(id = "trigger_downstream", retries = 2)
+ @Airflow.Task(id = "trigger_downstream", retries = 2)
public TriggerDagRun triggerDownstream() {
return new
TriggerDagRun("downstream_etl").config("wait_for_completion", true);
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
triggerDownstream();
@@ -2531,33 +2531,33 @@ class BuilderTest {
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TriggerDagRun;
- @Builder.Dag(id = "etl")
+ @Airflow.Dag(id = "etl")
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public long extract() {
return 42L;
}
- @Builder.Task
+ @Airflow.Task
public void load(long rows) {}
- @Builder.Task
+ @Airflow.Task
public void loadEmpty() {}
- @Builder.If(id = "has_rows")
+ @Airflow.If(id = "has_rows")
public boolean hasRows(long rows) {
return rows > 0;
}
- @Builder.Task(id = "trigger_downstream")
+ @Airflow.Task(id = "trigger_downstream")
public TriggerDagRun triggerDownstream() {
return new TriggerDagRun("downstream_etl");
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
var extracted = extract();
@@ -2600,16 +2600,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TriggerDagRun;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public TriggerDagRun triggerDownstream(long rows) {
return new TriggerDagRun("downstream_etl");
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2619,7 +2619,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Task method 'triggerDownstream' returns a TriggerDagRun, so it
runs when the Dag is " +
+ "@Airflow.Task method 'triggerDownstream' returns a TriggerDagRun, so it
runs when the Dag is " +
"built rather than when the task runs; it takes no parameters",
)
}
@@ -2631,16 +2631,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
+ import org.apache.airflow.sdk.Airflow;
import org.apache.airflow.sdk.TriggerDagRun;
- @Builder.Dag
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public TriggerDagRun triggerDownstream() throws java.io.IOException {
return new TriggerDagRun("downstream_etl");
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2650,7 +2650,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Task method 'triggerDownstream' returns a TriggerDagRun, so it
runs when the Dag is " +
+ "@Airflow.Task method 'triggerDownstream' returns a TriggerDagRun, so it
runs when the Dag is " +
"built, where a checked exception cannot be thrown; it must not throw:
java.io.IOException",
)
}
@@ -2662,15 +2662,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Switch
+ @Airflow.Switch
public String pickPath() {
return "handleLong";
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2680,7 +2680,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Switch method 'pickPath' returns java.lang.String, but a
switch returns the class of the task it chose",
+ "@Airflow.Switch method 'pickPath' returns java.lang.String, but a
switch returns the class of the task it chose",
)
}
@@ -2691,15 +2691,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Switch
+ @Airflow.Switch
public Class<String> pickPath() {
return String.class;
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2709,7 +2709,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.Switch method 'pickPath' returns
java.lang.Class<java.lang.String>, but a switch returns the class of the task
it chose",
+ "@Airflow.Switch method 'pickPath' returns
java.lang.Class<java.lang.String>, but a switch returns the class of the task
it chose",
)
}
@@ -2720,15 +2720,15 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.If
+ @Airflow.If
public String hasRows() {
return "yes";
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2738,7 +2738,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "@Builder.If method 'hasRows' returns java.lang.String, but a condition
returns boolean",
+ "@Airflow.If method 'hasRows' returns java.lang.String, but a condition
returns boolean",
)
}
@@ -2749,16 +2749,16 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
public class TestExample {
- @Builder.Task
- @Builder.If
+ @Airflow.Task
+ @Airflow.If
public boolean hasRows() {
return true;
}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {}
}
@@ -2768,7 +2768,7 @@ class BuilderTest {
assertThat(compilation).failed()
assertThat(compilation).hadErrorContaining(
- "Method 'hasRows' carries @Builder.Task, @Builder.If; a task is declared
by one of them alone",
+ "Method 'hasRows' carries @Airflow.Task, @Airflow.If; a task is declared
by one of them alone",
)
}
@@ -2779,13 +2779,13 @@ class BuilderTest {
compile(
"""
package org.apache.airflow.example;
- import org.apache.airflow.sdk.Builder;
- @Builder.Dag(id = "orders")
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(id = "orders")
public class TestExample {
- @Builder.Task
+ @Airflow.Task
public void t1() {}
- @Builder.Deps
+ @Airflow.Deps
static class Wiring implements TestExampleDeps {
void depends() {
t1();
diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts
index 0e247e65653..88226f5c57b 100644
--- a/java-sdk/sdk/build.gradle.kts
+++ b/java-sdk/sdk/build.gradle.kts
@@ -257,7 +257,7 @@ abstract class SyncDagSchemaTask : DefaultTask() {
// Generate the Dag-authoring DSL surface from the Dag serialization schema:
//
-// - org.apache.airflow.sdk.Builder and its nested Dag / Task annotations,
+// - org.apache.airflow.sdk.Airflow and its nested Dag / Task annotations,
// whose configuration attributes mirror the scalar keys of the schema's
// "dag" and "operator" definitions (the annotation processor lowers
// explicitly-set attributes into DagDef.config / TaskDef.config calls), and
@@ -508,7 +508,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
" /** ${f.doc} */\n val ${f.attribute}: ${f.attrType} =
${f.attrDefault},"
}
- outDir.resolve("org/apache/airflow/sdk").apply { mkdirs()
}.resolve("Builder.kt").writeText(
+ outDir.resolve("org/apache/airflow/sdk").apply { mkdirs()
}.resolve("Airflow.kt").writeText(
"""
|package org.apache.airflow.sdk
|
@@ -524,16 +524,16 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * Example:
| *
| * ```java
- | * @Builder.Dag(id = "my_pipeline", schedule = "@daily")
+ | * @Airflow.Dag(id = "my_pipeline", schedule = "@daily")
| * public class MyPipeline {
| *
- | * @Builder.Task(id = "extract", retries = 2)
+ | * @Airflow.Task(id = "extract", retries = 2)
| * public long extract(Client client) { ... }
| *
- | * @Builder.Task(id = "transform")
+ | * @Airflow.Task(id = "transform")
| * public long transform(Client client, long extracted) { ...
}
| *
- | * @Builder.Deps
+ | * @Airflow.Deps
| * static class Wiring implements MyPipelineDeps {
| * void depends() { transform(extract()); }
| * }
@@ -546,7 +546,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * the arguments bound at that call site take their place. Keyword
| * arguments bind by name instead through a single [TaskInput]
parameter.
| */
- |class Builder internal constructor() {
+ |class Airflow internal constructor() {
| /**
| * Annotation to automate a Dag-builder pattern.
| *
@@ -590,10 +590,10 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * [TaskRef], so each side is named where the Dag is wired:
| *
| * ```java
- | * @Builder.If(id = "has_rows")
+ | * @Airflow.If(id = "has_rows")
| * public boolean hasRows(long rows) { return rows > 0; }
| *
- | * @Builder.Deps
+ | * @Airflow.Deps
| * static class Wiring implements EtlDeps {
| * void depends() {
hasRows(extract()).Then(load()).Else(reportEmpty()); }
| * }
@@ -623,12 +623,12 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * [SwitchRef], so the cases are listed where the Dag is wired:
| *
| * ```java
- | * @Builder.Switch(id = "pick_path")
+ | * @Airflow.Switch(id = "pick_path")
| * public Class<? extends Task> pickPath(long rows) {
| * return rows > 1000 ? EtlBuilder.HandleLong.class :
EtlBuilder.HandleShort.class;
| * }
| *
- | * @Builder.Deps
+ | * @Airflow.Deps
| * static class Wiring implements EtlDeps {
| * void depends() {
pickPath(extract()).Case(handleLong()).Case(handleShort()); }
| * }
@@ -658,7 +658,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * to rather than an id it owns:
| *
| * ```java
- | * @Builder.TaskHandler(dag = "etl", task = "score")
+ | * @Airflow.TaskHandler(dag = "etl", task = "score")
| * public long score(Client client, long rows) { ... }
| * ```
| *
@@ -687,7 +687,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * wire an ordering-only one:
| *
| * ```java
- | * @Builder.Deps
+ | * @Airflow.Deps
| * static class Wiring implements EtlPipelineDeps {
| * void depends() {
| * var rows = extract();
@@ -716,14 +716,14 @@ abstract class GenerateDagDslTask : DefaultTask() {
| * task `Staging.stage`:
| *
| * ```java
- | * @Builder.TaskGroup
+ | * @Airflow.TaskGroup
| * static class Staging {
- | * @Builder.Task
+ | * @Airflow.Task
| * public long stage(long rows) { ... }
| *
- | * @Builder.TaskGroup(id = "checks")
+ | * @Airflow.TaskGroup(id = "checks")
| * static class Checks {
- | * @Builder.Task
+ | * @Airflow.Task
| * public void nulls(long staged) { ... }
| * }
| * }
@@ -770,7 +770,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
|/**
| * Configuration keys accepted by `DagDef.config` and
`TaskDef.config`,
| * keyed by Dag serialization schema property name. Public so
that the
- | * annotation processor can lower `@Builder.Dag` / `@Builder.Task`
+ | * annotation processor can lower `@Airflow.Dag` / `@Airflow.Task`
| * attributes onto the same tables; not user-facing API.
| */
|object SchemaFields {
@@ -822,7 +822,7 @@ val syncDagSchema by
tasks.registering(SyncDagSchemaTask::class) {
tasks.register<GenerateDagDslTask>("generateDagDsl") {
dependsOn(syncDagSchema)
- description = "Generate the Builder.Dag/Builder.Task annotations and
SchemaFields from the Dag serialization schema"
+ description = "Generate the Airflow.Dag/Airflow.Task annotations and
SchemaFields from the Dag serialization schema"
schemaFile = dagSchemaInput
targetDirectory = dagDslDir
}
diff --git a/java-sdk/sdk/module.md b/java-sdk/sdk/module.md
index 2cc58f52b4d..4e9f368e475 100644
--- a/java-sdk/sdk/module.md
+++ b/java-sdk/sdk/module.md
@@ -60,7 +60,7 @@ meaning of each dimension is defined in the
| 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 † | ✗ | – | |
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Arg.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Arg.kt
index bff37ad02ef..765cfc88822 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Arg.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Arg.kt
@@ -34,7 +34,7 @@ sealed class Arg<T> {
* Wraps an inline constant as a task argument, passed to the task as a
* constant and creating no dependency edge.
*
- * [Deps.lit] is the spelling a [Builder.Deps] class uses; this is the same
+ * [Deps.lit] is the spelling an [Airflow.Deps] class uses; this is the
same
* call for code with no `Deps` in scope.
*
* @param value Constant to bind; may be null for a nullable parameter.
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt
index 5e817c05e93..232bf3b723d 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt
@@ -115,25 +115,25 @@ class Bundle(
* Registers what an annotated class holds, read from the class itself so
* there is no second name to keep in sync.
*
- * A [Builder.Dag] class contributes the Dag its generated builder builds; a
- * class of [Builder.TaskHandler] methods contributes each handler, bound to
+ * An [Airflow.Dag] class contributes the Dag its generated builder builds; a
+ * class of [Airflow.TaskHandler] methods contributes each handler, bound to
* the Dag the Python file owns. A class can carry both.
*
- * @param annotated A class carrying [Builder.Dag] or [Builder.TaskHandler].
+ * @param annotated A class carrying [Airflow.Dag] or [Airflow.TaskHandler].
* @return This bundle, for chaining.
* @throws IllegalArgumentException if the class has no generated code,
* because annotation processing did not run over it, or if the Dag's
- * wiring is invalid, such as a task the `@Builder.Deps` class did not
+ * wiring is invalid, such as a task the `@Airflow.Deps` class did not
* call.
*/
fun register(annotated: Class<*>): Bundle {
checkOpen()
- val dag = annotated.getAnnotation(Builder.Dag::class.java)
+ val dag = annotated.getAnnotation(Airflow.Dag::class.java)
if (dag != null) {
val builder = generated(builderName(annotated.packageName,
annotated.simpleName, dag.to), annotated, "builder")
register(invokeGenerated(builder.getMethod("build")) as DagDef)
}
- if (dag == null || annotated.declaredMethods.any {
it.isAnnotationPresent(Builder.TaskHandler::class.java) }) {
+ if (dag == null || annotated.declaredMethods.any {
it.isAnnotationPresent(Airflow.TaskHandler::class.java) }) {
val registrar = generated(registrarName(annotated.name), annotated,
"registrar")
invokeGenerated(registrar.getMethod("registerInto", Bundle::class.java),
this)
}
@@ -184,8 +184,8 @@ class Bundle(
Class.forName(name, true, from.classLoader)
} catch (e: ClassNotFoundException) {
throw IllegalArgumentException(
- "No generated $what $name for ${from.name}; does it carry @Builder.Dag
or " +
- "@Builder.TaskHandler, and is airflow-sdk-processor on the
annotationProcessor path?",
+ "No generated $what $name for ${from.name}; does it carry @Airflow.Dag
or " +
+ "@Airflow.TaskHandler, and is airflow-sdk-processor on the
annotationProcessor path?",
e,
)
}
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
index 4b275c4b7ea..06630cc89f4 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
@@ -32,7 +32,7 @@ import kotlin.Throws
*
* Create a [DagDef] directly and register [TaskDef]s with [addTask].
*
- * The [Builder.Dag] annotation should generally be preferred in user code,
+ * The [Airflow.Dag] annotation should generally be preferred in user code,
* where the annotation processor generates the wiring for you. Only use this
* class directly if you need to do low-level plumbing:
*
@@ -49,7 +49,7 @@ import kotlin.Throws
*
* @param id Dag identifier; must be unique within a [Bundle].
*
- * @see Builder.Dag
+ * @see Airflow.Dag
*/
class DagDef(
val id: String,
@@ -427,7 +427,7 @@ private val EMPTY_GROUP_EDGES = GroupEdges()
* declared input cannot be bound, so that a mis-declared input fails while
* the [Bundle] is built rather than mid-run.
*
- * @see Builder.Task
+ * @see Airflow.Task
*/
class TaskDef(
val id: String,
@@ -500,7 +500,7 @@ class TaskDef(
/**
* A single unit of work executed by Airflow.
*
- * Prefer using the [Builder.Task] annotation with [Builder.Dag] to have the
+ * Prefer using the [Airflow.Task] annotation with [Airflow.Dag] to have the
* annotation processor generate an implementation for you. Only use this
* interface if you need to do low-level plumbing.
*
@@ -511,8 +511,8 @@ class TaskDef(
* Implement [InputTask] instead for a task the Python Dag file calls with
* TaskFlow arguments; the SDK then resolves those arguments and injects them.
*
- * @see Builder.Dag
- * @see Builder.Task
+ * @see Airflow.Dag
+ * @see Airflow.Task
* @see InputTask
*/
interface Task {
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
index 01b26177cd6..1cf90491841 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
@@ -25,7 +25,7 @@ import org.apache.airflow.sdk.internal.Refs
* Vocabulary for declaring a Dag's task graph in Java, and the base of every
* generated `<Dag>Deps` wiring view.
*
- * A [Builder.Deps] class inherits [lit] and [Flow] by simple name, so it needs
+ * An [Airflow.Deps] class inherits [lit] and [Flow] by simple name, so it
needs
* no import and `Flow` does not collide with `java.util.concurrent.Flow`.
*/
interface Deps {
@@ -99,7 +99,7 @@ interface Deps {
* One task group of the Dag being wired: a point in the flow, and the
* namespace of the tasks and groups declared inside it.
*
- * The generated wiring view nests one of these per [Builder.TaskGroup]
+ * The generated wiring view nests one of these per [Airflow.TaskGroup]
* class, so a group is reached by calling it and its contents by calling on
* through:
*
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskInput.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskInput.kt
index 8fd78fa632c..a29372bc08d 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskInput.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskInput.kt
@@ -36,12 +36,12 @@ package org.apache.airflow.sdk
* public double threshold; // binds threshold
* }
*
- * @Builder.TaskHandler(dag = "etl", task = "score")
+ * @Airflow.TaskHandler(dag = "etl", task = "score")
* public Result score(Client client, ScoreInput input) { ... }
* ```
*
* A `TaskInput` binds the same way whichever authoring API declares it: as a
- * `@Builder.TaskHandler` parameter, as above, or as the input type of an
+ * `@Airflow.TaskHandler` parameter, as above, or as the input type of an
* [InputTask]. A task method may declare at most one `TaskInput` parameter
* and, if it does, no other data parameters — the `TaskInput` owns the whole
named-argument
* surface, so field names and flat positions cannot shift each other.
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt
index c78d1a9fd5f..15a6a5b3d0d 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt
@@ -60,7 +60,7 @@ private val TRIGGER_FIELDS: Map<String, Field> =
* .config("conf", Map.of("rows", 2)));
* ```
*
- * In an annotated Dag, a `@Builder.Task` method returns one instead; the
+ * In an annotated Dag, an `@Airflow.Task` method returns one instead; the
* method runs when the Dag is built, not when the task runs.
*
* The task runs no Java code and takes no arguments. It pushes the triggered
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
index e3920f2b67f..25b5f021e47 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
@@ -68,7 +68,7 @@ object ArgValues {
* claims.
*
* The single populator behind both authoring APIs — the annotation processor
- * emits a call to it for a `@Builder.Task` [TaskInput] parameter, and
+ * emits a call to it for an `@Airflow.Task` [TaskInput] parameter, and
* [org.apache.airflow.sdk.InputTask] calls it before handing the input to a
* task written against the interface.
*
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt
index 5b813b5c733..0afea428eed 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt
@@ -25,7 +25,7 @@ import java.time.OffsetDateTime
/**
* Value shape of one Dag serialization schema key. Public so that the
- * annotation processor can lower `@Builder.Dag` / `@Builder.Task` attributes;
+ * annotation processor can lower `@Airflow.Dag` / `@Airflow.Task` attributes;
* not user-facing API.
*/
enum class FieldType {
@@ -49,7 +49,7 @@ private val DAG_RUN_STATE_NAMES = listOf("queued", "running",
"success", "failed
/**
* One configuration key from the Dag serialization schema. Public so that the
- * annotation processor can lower `@Builder.Dag` / `@Builder.Task` attributes;
+ * annotation processor can lower `@Airflow.Dag` / `@Airflow.Task` attributes;
* not user-facing API.
*
* @property key Schema property name, e.g. `retry_delay`.
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/InputTypes.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/InputTypes.kt
index fad312ffcf3..0d56fd35a4c 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/InputTypes.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/InputTypes.kt
@@ -50,7 +50,7 @@ internal fun resolveInputType(taskClass: Class<*>): Class<out
TaskInput> =
* is being built rather than mid-run. A plain [Task] declares no input and
passes.
*
* The annotation processor enforces the same rules at compile time for the
- * [TaskInput] a `@Builder.Task` method declares.
+ * [TaskInput] an `@Airflow.Task` method declares.
*
* @throws IllegalArgumentException if the declared input type is unresolvable
* or cannot be populated.
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
index 761d5b8f282..c668020b21d 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
@@ -28,7 +28,7 @@ import org.apache.airflow.sdk.TaskRef
/**
* @suppress
*
- * The recorder behind a `@Builder.Deps` wiring class. Public so that
+ * The recorder behind an `@Airflow.Deps` wiring class. Public so that
* processor-generated wiring views can call it; not user-facing API.
*
* A wiring view's methods are `default` methods on an interface, so they hold
@@ -75,7 +75,7 @@ object Refs {
val missing = taskIds.filterNot { it in dag.tasks }
require(missing.isEmpty()) {
"Wiring for Dag '${dag.id}' did not register task(s)
${missing.joinToString { "'$it'" }}: " +
- "every @Builder.Task method must be called in the @Builder.Deps class"
+ "every @Airflow.Task method must be called in the @Airflow.Deps class"
}
return dag
}
@@ -126,7 +126,7 @@ object Refs {
}
val active =
checkNotNull(recording.get()) {
- "Task '${def.id}' was wired outside a @Builder.Deps class; the wiring
view's methods " +
+ "Task '${def.id}' was wired outside an @Airflow.Deps class; the wiring
view's methods " +
"only record while the generated builder is running depends()"
}
active.byTaskId[def.id]?.let { existing ->
@@ -156,7 +156,7 @@ object Refs {
fun group(id: String): TaskGroupRef {
val active =
checkNotNull(recording.get()) {
- "Task group '$id' was looked up outside a @Builder.Deps class"
+ "Task group '$id' was looked up outside an @Airflow.Deps class"
}
return requireNotNull(active.dag.groups[id]) {
"Dag '${active.dag.id}' has no task group '$id'"
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Registrar.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Registrar.kt
index 30edadb3060..60a4d715c67 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Registrar.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Registrar.kt
@@ -22,7 +22,7 @@ package org.apache.airflow.sdk.internal
/**
* @suppress
*
- * Names the registrar generated for a class of `@Builder.TaskHandler`
+ * Names the registrar generated for a class of `@Airflow.TaskHandler`
* methods: a top-level class in the handler class's package, named after that
* class and any class enclosing it, so `Outer.Inner` is served by
* `Outer_InnerHandlers`.
@@ -38,7 +38,7 @@ fun registrarName(binaryName: String): String =
"${binaryName.replace('$', '_')}
/**
* @suppress
*
- * Names the builder generated for a `@Builder.Dag` class: a top-level class
+ * Names the builder generated for an `@Airflow.Dag` class: a top-level class
* in the Dag class's package, named by the annotation's `to` or, when that is
* blank, `<Class>Builder`.
*
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt
index 4c645aab89f..2676b070b3f 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt
@@ -35,17 +35,17 @@ internal class BundleTest {
class Nested
/** A Dag class that the processor generated [WiredDagBuilder] for. */
- @Builder.Dag(id = "wired")
+ @Airflow.Dag(id = "wired")
class WiredDag
/** A Dag class whose generated [BrokenDagBuilder] fails, as an invalid
wiring would. */
- @Builder.Dag
+ @Airflow.Dag
class BrokenDag
/** A Dag class that also holds a task handler, so the processor generated
both. */
- @Builder.Dag(id = "mixed")
+ @Airflow.Dag(id = "mixed")
class MixedDag {
- @Builder.TaskHandler(dag = "etl", task = "score")
+ @Airflow.TaskHandler(dag = "etl", task = "score")
fun score() = Unit
}
@@ -150,7 +150,7 @@ internal class BundleTest {
}
@Test
- @DisplayName("Should register the Dag a @Builder.Dag class's generated
builder builds")
+ @DisplayName("Should register the Dag an @Airflow.Dag class's generated
builder builds")
fun shouldRegisterDagFromBuilderClass() {
val bundle = Bundle().register(WiredDag::class.java)
@@ -209,7 +209,7 @@ internal class BundleTest {
Assertions.assertEquals(
"No generated registrar org.apache.airflow.sdk.BundleTest_NoOpHandlers
for " +
- "${NoOp::class.java.name}; does it carry @Builder.Dag or
@Builder.TaskHandler, " +
+ "${NoOp::class.java.name}; does it carry @Airflow.Dag or
@Airflow.TaskHandler, " +
"and is airflow-sdk-processor on the annotationProcessor path?",
error.message,
)
diff --git
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
index cafbd5e40d2..bcb254547be 100644
---
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
+++
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
@@ -54,7 +54,7 @@ private class NoopArgTask : Task {
) = Unit
}
-/** Resolution of the inputs a `@Builder.Deps` class recorded, without runtime
bindings. */
+/** Resolution of the inputs an `@Airflow.Deps` class recorded, without
runtime bindings. */
internal class ArgValuesTest {
/** Upstream task ids read through the transport, in arrival order. */
private val pulls = java.util.concurrent.CopyOnWriteArrayList<String>()
diff --git
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
index c764585e409..ef1748e8472 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
@@ -125,7 +125,7 @@ internal class RefsTest {
assertEquals(
"Wiring for Dag 'd' did not register task(s) 'x', 'y': " +
- "every @Builder.Task method must be called in the @Builder.Deps class",
+ "every @Airflow.Task method must be called in the @Airflow.Deps class",
error.message,
)
}
@@ -139,7 +139,7 @@ internal class RefsTest {
}
assertEquals(
- "Task 't' was wired outside a @Builder.Deps class; the wiring view's
methods " +
+ "Task 't' was wired outside an @Airflow.Deps class; the wiring view's
methods " +
"only record while the generated builder is running depends()",
error.message,
)
diff --git
a/kubernetes-tests/lang_sdk/java_example/src/java/org/apache/airflow/k8sexample/MixedLanguageExample.java
b/kubernetes-tests/lang_sdk/java_example/src/java/org/apache/airflow/k8sexample/MixedLanguageExample.java
index 1a35838c82b..64251d1defa 100644
---
a/kubernetes-tests/lang_sdk/java_example/src/java/org/apache/airflow/k8sexample/MixedLanguageExample.java
+++
b/kubernetes-tests/lang_sdk/java_example/src/java/org/apache/airflow/k8sexample/MixedLanguageExample.java
@@ -32,13 +32,13 @@ import org.slf4j.LoggerFactory;
public class MixedLanguageExample {
private static final Logger logger =
LoggerFactory.getLogger(MixedLanguageExample.class);
- @Builder.TaskHandler(dag = "lang_sdk_mixed_language", task = "java_extract")
+ @Airflow.TaskHandler(dag = "lang_sdk_mixed_language", task = "java_extract")
public long extract(Client client) {
logger.info("java_extract running");
return new Date().getTime();
}
- @Builder.TaskHandler(dag = "lang_sdk_mixed_language", task =
"java_transform")
+ @Airflow.TaskHandler(dag = "lang_sdk_mixed_language", task =
"java_transform")
public void transform(Client client) {
var variable = client.getVariable("my_variable");
logger.info("java_transform obtained variable {}", variable);
diff --git a/task-sdk/docs/lang-sdk-spec.rst b/task-sdk/docs/lang-sdk-spec.rst
index a260b7a56fc..dfcf76c1bb4 100644
--- a/task-sdk/docs/lang-sdk-spec.rst
+++ b/task-sdk/docs/lang-sdk-spec.rst
@@ -105,11 +105,11 @@ Per-SDK spelling
- an ``async`` function
* - ``TaskHandler``
- ``airflow.TaskHandler(dagId, taskId, fn)``
- - ``@Builder.TaskHandler(dagId, taskId)``
+ - ``@Airflow.TaskHandler(dagId, taskId)``
- ``new TaskHandler(dagId, taskId, fn)``
* - ``Dag``
- ``airflow.Dag(spec)``
- - ``@Builder.Dag(id, ...)``, or ``new Dag(id)``
+ - ``@Airflow.Dag(id, ...)``, or ``new Dag(id)``
- ``new Dag(id)``
* - ``dag``
- ``*airflow.DagRef``
@@ -117,7 +117,7 @@ Per-SDK spelling
- ``Dag``
* - ``dag.Task``
- ``dag.Task(fn, opts...)``
- - ``@Builder.Task(id)``, or ``dag.task(id, cls)``
+ - ``@Airflow.Task(id)``, or ``dag.task(id, cls)``
- ``dag.task(id, fn)``
* - data edge
- ``airflow.Inputs(refs...)``
