Skip to content

Commit b7a3d3d

Browse files
committed
Java SDK: Declare a Dag's task graph with @Builder.Deps
An annotated Dag class could name its tasks but not say how they connect, so a Dag whose annotations carried everything else still needed a Python Dag file for its graph. Declaring the graph by calling the tasks lets javac check that each task is fed something its parameter accepts. Bundle.register(Class) now also takes the @Builder.Dag class itself and registers the Dag its generated builder builds, so there is still only one name to keep in sync.
1 parent 8126c9e commit b7a3d3d

16 files changed

Lines changed: 1091 additions & 117 deletions

File tree

‎airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -314,6 +314,9 @@ Annotate a plain Java class and let the SDK generate the boilerplate at compile
314314
- Marks a method as a task of a Java-owned Dag. If ``id`` is omitted the method name is
315315
used. Further attributes (``retries``, ``queue``, ``retryDelay``, …) mirror the Dag
316316
serialization schema; only attributes written explicitly are applied.
317+
* - ``@Builder.Deps``
318+
- Marks the nested class that declares the task graph in Java, TaskFlow-style. Only needed
319+
for a Dag that has no Python stub file. See :ref:`java-sdk/native-dags`.
317320
* - ``TaskInput`` / ``@ArgName("...")``
318321
- Marks a class as a task's input, so keyword arguments bind by name instead of by position:
319322
each public field receives the argument whose name matches it, ignoring case and
@@ -569,6 +572,79 @@ Edges are checked when the Dag is registered with a ``Bundle``: an upstream that
569572
Dag, or to no Dag, and a cycle anywhere in the graph both fail there rather than at the first task
570573
run.
571574

575+
Wiring the graph with ``@Builder.Deps``
576+
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
577+
578+
For a Dag written with annotations, the graph is declared by a nested ``@Builder.Deps`` class. The
579+
annotation processor generates a ``<ClassName>Deps`` interface — the *wiring view* — with one method
580+
per ``@Builder.Task`` method: the injected ``Client`` and ``Context`` parameters are dropped, each
581+
data parameter becomes an ``Arg<T>``, and the return value becomes a ``TaskRef<T>``. Calling a view
582+
method registers its task, and passing the handle one returned into another feeds the upstream's
583+
output into the downstream's parameter *and* wires the data edge. The call graph is the task graph,
584+
and ``javac`` type-checks it.
585+
586+
Declare the wiring class as a ``static`` nested class of the Dag class that ``implements`` the
587+
generated view, with a no-argument ``depends()`` method:
588+
589+
.. code-block:: java
590+
591+
@Builder.Dag(
592+
id = "java_etl",
593+
schedule = "@daily",
594+
description = "Pure-Java Dag, no Python stub file",
595+
tags = {"example", "java-sdk"})
596+
public class EtlPipeline {
597+
598+
@Builder.Task(id = "extract", retries = 2)
599+
public long extract() {
600+
return 42L;
601+
}
602+
603+
@Builder.Task(id = "transform")
604+
public long transform(long extracted, double factor) {
605+
return (long) (extracted * factor);
606+
}
607+
608+
@Builder.Task(id = "load")
609+
public void load(long transformed) {
610+
// implement task logic
611+
}
612+
613+
@Builder.Task(id = "audit")
614+
public void audit() {
615+
// side effect only, no data in or out
616+
}
617+
618+
@Builder.Deps
619+
static class Wiring implements EtlPipelineDeps {
620+
void depends() {
621+
var rows = extract();
622+
load(transform(rows, lit(0.9)));
623+
rows.then(audit()); // ordering-only edge: extract >> audit
624+
}
625+
}
626+
}
627+
628+
Every ``@Builder.Task`` method must be called in the wiring class; a task the wiring missed fails at
629+
Dag-parse time. ``lit(...)`` wires an inline constant where no upstream feeds a parameter — a bare
630+
``double`` cannot be an ``Arg``, so a constant is wrapped. A view method that takes no arguments
631+
returns the same handle every time, so it names one node wherever it appears; one that takes
632+
arguments is called once, and the wiring fails if it is called again with arguments, so hold its
633+
handle in a local and reuse that.
634+
635+
``then`` and ``Flow.of`` work here exactly as they do on the interface surface; inside the wiring
636+
class ``Flow`` is inherited by simple name, so it needs no import and never collides with
637+
``java.util.concurrent.Flow``.
638+
639+
The wiring class is optional — a Dag class without one registers every task with no Java-side edges,
640+
which is the shape for stub-backed tasks whose graph the Python Dag file owns.
641+
642+
.. note::
643+
644+
Runtime argument bindings win over Java-declared wiring. When the supervisor delivers bindings
645+
for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's position is what the
646+
task receives. Wired inputs are the fallback, which is what a native Java Dag always uses.
647+
572648
Configuration attributes
573649
~~~~~~~~~~~~~~~~~~~~~~~~
574650

‎java-sdk/README.md‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -727,9 +727,9 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
727727
- The annotation processor (`BuilderProcessor.kt`) uses `kapt`. The `Builder`
728728
class holding the `@Builder.Dag` / `@Builder.Task` annotations is generated
729729
from the Dag serialization schema by `:sdk:generateDagDsl` (vendored at
730-
`sdk/schema/dag-schema.json`). The `Arg`/`TaskRef` and `Deps`/`Flow` graph
731-
types are hand-written next to the rest of the public surface in
732-
`sdk/src/main/kotlin/org/apache/airflow/sdk/`. When adding annotation
730+
`sdk/schema/dag-schema.json`), `@Builder.Deps` included. The `Arg`/`TaskRef`
731+
and `Deps`/`Flow` graph types are hand-written next to the rest of the public
732+
surface in `sdk/src/main/kotlin/org/apache/airflow/sdk/`. When adding annotation
733733
behaviour, handle it in `BuilderProcessor.kt` and add a golden-output test in
734734
`processor/src/test/kotlin/`.
735735
- The Python coordinator subclasses `SubprocessCoordinator`. Do not reach into

‎java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,12 +21,15 @@
2121

2222
import org.apache.airflow.sdk.*;
2323

24+
// One bundle serves every surface: Dags built in Java, and the handler classes
25+
// whose Dags the Python file owns.
2426
public class ExampleBundleBuilder {
2527
public static Bundle build() {
2628
return new Bundle()
2729
.register(InterfaceExampleBuilder.build())
2830
.register(AnnotationExample.class)
2931
.register(XComCastingExample.class)
32+
.register(org.apache.airflow.example.nativedag.AnnotationExample.class)
3033
.register(org.apache.airflow.example.nativedag.InterfaceExample.build());
3134
}
3235

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
// "native" is a Java keyword, so the native-Dag examples live in "nativedag".
21+
package org.apache.airflow.example.nativedag;
22+
23+
import static java.lang.System.Logger.Level.INFO;
24+
25+
import org.apache.airflow.sdk.*;
26+
27+
// A Dag defined entirely in Java, annotation-style: no Python stub file
28+
// describes it. The @Builder.Dag/@Builder.Task attributes carry the
29+
// configuration and the @Builder.Deps class declares the graph -- the single
30+
// place dependencies are defined.
31+
@Builder.Dag(
32+
id = "java_native_annotation_example",
33+
description = "Pure-Java Dag authored with annotations, without a Python stub file",
34+
schedule = "@daily",
35+
startDate = "2026-01-01T00:00:00Z",
36+
catchup = false,
37+
tags = {"example", "java-sdk"})
38+
public class AnnotationExample {
39+
private static final System.Logger log = System.getLogger(AnnotationExample.class.getName());
40+
41+
@Builder.Task(id = "extract", retries = 2)
42+
public long extract() {
43+
log.log(INFO, "Extracting a value");
44+
return 42L;
45+
}
46+
47+
@Builder.Task(id = "transform")
48+
public long transform(long extracted, double factor) {
49+
log.log(INFO, "Transforming {0} by {1}", extracted, factor);
50+
return (long) (extracted * factor);
51+
}
52+
53+
@Builder.Task(id = "load")
54+
public void load(long transformed) {
55+
log.log(INFO, "Loaded {0}", transformed);
56+
}
57+
58+
@Builder.Task(id = "audit")
59+
public void audit() {
60+
log.log(INFO, "Audited the run");
61+
}
62+
63+
// Implements the generated wiring view, so javac type-checks the graph:
64+
// extract() yields a TaskRef<Long>, which is an Arg<Long> for transform.
65+
@Builder.Deps
66+
static class Wiring implements AnnotationExampleDeps {
67+
void depends() {
68+
var extracted = extract();
69+
load(transform(extracted, lit(1.5)));
70+
// Ordering-only edge: audit runs after extract, with no data flowing.
71+
extracted.then(audit());
72+
}
73+
}
74+
}

0 commit comments

Comments
 (0)