Java SDK: Declare a Dag's task graph with @Builder.Deps

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

@@ -314,6 +314,9 @@ Annotate a plain Java class and let the SDK generate the boilerplate at compile
      - 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``
+     - 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("...")``
      - Marks a class as a task's input, so keyword arguments bind by name instead of by position:
        each public field receives the argument whose name matches it, ignoring case and
@@ -574,6 +577,80 @@ 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``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+For a Dag written with annotations, the graph is declared by a nested ``@Builder.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
+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,
+and ``javac`` type-checks it.
+
+Declare the wiring class as a ``static`` nested class of the Dag class that ``implements`` the
+generated view, with a no-argument ``depends()`` method:
+
+.. code-block:: java
+
+    @Builder.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)
+      public long extract() {
+        return 42L;
+      }
+
+      @Builder.Task(id = "transform")
+      public long transform(long extracted, double factor) {
+        return (long) (extracted * factor);
+      }
+
+      @Builder.Task(id = "load")
+      public void load(long transformed) {
+        // implement task logic
+      }
+
+      @Builder.Task(id = "audit")
+      public void audit() {
+        // side effect only, no data in or out
+      }
+
+      @Builder.Deps
+      static class Wiring implements EtlPipelineDeps {
+        void depends() {
+          var rows = extract();
+          load(transform(rows, lit(0.9)));
+          rows.before(audit()); // ordering-only edge: audit waits for extract
+        }
+      }
+    }
+
+Every ``@Builder.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.  A view method that takes no arguments
+returns the same handle every time, so it names one node wherever it appears; one that takes
+arguments is called once, and the wiring fails if it is called again with arguments, so hold its
+handle in a local and reuse that.
+
+``before``, ``after`` and ``Flow.of`` work here exactly as they do on the interface surface; inside
+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
+class that supplies only task bodies, for a Dag a Python file declares, carries
+``@Builder.TaskHandler`` instead and contributes no Dag.
+
+.. note::
+
+   Runtime argument bindings win over Java-declared wiring.  When the supervisor delivers bindings
+   for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's position is what the
+   task receives.  Wired inputs are the fallback, which is what a native Java Dag always uses.
+
 Configuration attributes
 ~~~~~~~~~~~~~~~~~~~~~~~~
 

java-sdk/README.md

@@ -729,9 +729,9 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
 - The annotation processor (`BuilderProcessor.kt`) uses `kapt`. The `Builder`
   class holding the `@Builder.Dag` / `@Builder.Task` annotations is generated
   from the Dag serialization schema by `:sdk:generateDagDsl` (vendored at
-  `sdk/schema/dag-schema.json`). 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
+  `sdk/schema/dag-schema.json`), `@Builder.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
   `processor/src/test/kotlin/`.
 - The Python coordinator subclasses `SubprocessCoordinator`. Do not reach into

java-sdk/adr/0002-native-dag-interface.md

@@ -65,7 +65,8 @@ public class EtlPipeline { // extends nothing of ours; your own base class stays
   public void audit(Client client) { /* side effect only, no data in or out */ }
 
   @Builder.Task(id = "notify")
-  public void notify(Client client) { /* side effect only, no data in or out */ }
+  // "notify" is Object.notify, so the method takes another name and keeps the task id.
+  public void alert(Client client) { /* side effect only, no data in or out */ }
 
   @Builder.Deps
   static class Wiring implements EtlPipelineDeps {
@@ -77,7 +78,7 @@ public class EtlPipeline { // extends nothing of ours; your own base class stays
 
       // non-TaskFlow (ordering-only) edges: sequence with no data flowing
       rows.then(audit());              // extract >> audit
-      Flow.of(loaded, audit()).then(notify()); // [load, audit] >> notify
+      Flow.of(loaded, audit()).then(alert()); // [load, audit] >> notify
     }
   }
 
@@ -101,7 +102,7 @@ interface EtlPipelineDeps extends Deps { // Deps: the shared base that nests Flo
   default TaskRef<Long> transform(Arg<Long> extracted, Arg<Double> threshold) { return Flow.call("transform", extracted, threshold); }
   default TaskRef<Void> load(Arg<Long> transformed) { return Flow.call("load", transformed); }
   default TaskRef<Void> audit() { return Flow.node("audit"); }
-  default TaskRef<Void> notify() { return Flow.node("notify"); }
+  default TaskRef<Void> alert() { return Flow.node("notify"); }
 }
 ```
 

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

@@ -21,12 +21,15 @@
 
 import org.apache.airflow.sdk.*;
 
+// One bundle serves every surface: Dags built in Java, and the handler classes
+// whose Dags the Python file owns.
 public class ExampleBundleBuilder {
   public static Bundle build() {
     return new Bundle()
         .register(InterfaceExampleBuilder.build())
         .register(AnnotationExample.class)
         .register(XComCastingExample.class)
+        .register(org.apache.airflow.example.nativedag.AnnotationExample.class)
         .register(org.apache.airflow.example.nativedag.InterfaceExample.build());
   }
 

java-sdk/example/src/java/org/apache/airflow/example/nativedag/AnnotationExample.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// "native" is a Java keyword, so the native-Dag examples live in "nativedag".
+package org.apache.airflow.example.nativedag;
+
+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
+// declares the graph -- the single place dependencies are defined.
+@Builder.Dag(
+    id = "java_native_annotation_example",
+    description = "Pure-Java Dag authored with annotations",
+    schedule = "@daily",
+    startDate = "2026-01-01T00:00:00Z",
+    catchup = false,
+    tags = {"example", "java-sdk"})
+public class AnnotationExample {
+  private static final System.Logger log = System.getLogger(AnnotationExample.class.getName());
+
+  @Builder.Task(id = "extract", retries = 2)
+  public long extract() {
+    log.log(INFO, "Extracting a value");
+    return 42L;
+  }
+
+  @Builder.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")
+  public void load(long transformed) {
+    log.log(INFO, "Loaded {0}", transformed);
+  }
+
+  @Builder.Task(id = "audit")
+  public void audit() {
+    log.log(INFO, "Audited the run");
+  }
+
+  // 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
+  static class Wiring implements AnnotationExampleDeps {
+    void depends() {
+      var extracted = extract();
+      load(transform(extracted, lit(1.5)));
+      // Ordering-only edge: audit runs after extract, with no data flowing.
+      extracted.before(audit());
+    }
+  }
+}