nprasad2077/airflow
1
0 Fork 0
mirror of https://github.com/apache/airflow.git synced 2026-11-04 07:12:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Jason(Zhe-You) Liu authored and GitHub committed 2026-10-13 18:57:47 +01:00
1 parent a3c8ee41d0
commit d479721a4a
30 files changed
+555 -555
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

java-sdk/README.md
+8 -8
View File
@@ -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
java-sdk/capabilities.yaml
+1 -1
View File
@@ -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:
java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
+5 -5
View File
@@ -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 {
java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
+8 -8
View File
@@ -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);
java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java
+15 -15
View File
@@ -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.
@Builder.Dag(
@Airflow.Dag(
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();
java-sdk/processor/build.gradle.kts
+1 -1
View File
@@ -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."
}
}
}
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt → java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
+60 -60
View File
@@ -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 {
java-sdk/processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor
+1 -1
View File
@@ -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
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt → java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
+341 -341
View File
File diff suppressed because it is too large. Load diff
java-sdk/sdk/build.gradle.kts
+19 -19
View File
@@ -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
}
java-sdk/sdk/module.md
+1 -1
View File
@@ -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 † | ✗ | – | |
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Arg.kt
+1 -1
View File
@@ -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.
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt
+8 -8
View File
@@ -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,
)
}
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
+6 -6
View File
@@ -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 {
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
+2 -2
View File
@@ -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:
*
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskInput.kt
+2 -2
View File
@@ -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.
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt
+1 -1
View File
@@ -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
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
+1 -1
View File
@@ -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.
*
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt
+2 -2
View File
@@ -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`.
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/InputTypes.kt
+1 -1
View File
@@ -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.
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
+4 -4
View File
@@ -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'"
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Registrar.kt
+2 -2
View File
@@ -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`.
*
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt
+6 -6
View File
@@ -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,
)
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
+1 -1
View File
@@ -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>()
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
+2 -2
View File
@@ -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,
)