Java SDK: Trigger another Dag's run from a native Java Dag

.pre-commit-config.yaml

@@ -444,7 +444,7 @@ repos:
           ^java-sdk/scripts/ci/prek/check_serialization_conformance\.py$|
           ^java-sdk/sdk/schema/dag-schema\.json$|
           ^java-sdk/sdk/build\.gradle\.kts$|
-          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/(Switch|Bundle|Condition|DagDef|Decider|Deps|Arg|TaskGroupRef)\.kt$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/(Switch|Bundle|Condition|DagDef|Decider|Deps|Arg|TaskGroupRef|TriggerDagRun)\.kt$|
           ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde\.kt$|
           ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/(Refs|Fields)\.kt$|
           ^java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/.*$|

java-sdk/README.md

@@ -620,8 +620,8 @@ prek hook regenerate it.
 | state: `success` | MUST | ✓ | 3.3 |  |
 | state: `failed` | MUST | ✓ | 3.3 |  |
 | state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask |
-| state: `skipped` | SHOULD | ✗ | – | runtime does not emit TaskState skipped yet |
-| state: `deferred` | MAY | ✗ | – | runtime does not emit DeferTask yet |
+| state: `skipped` | SHOULD | ✓ | 3.4 | a TriggerDagRun task that skips an existing run |
+| state: `deferred` | MAY | ✓ | 3.4 | a TriggerDagRun task that waits with deferrable |
 | state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet |
 | state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet |
 | state: `removed` | MAY | ✓ | 3.3 |  |

java-sdk/capabilities.yaml

@@ -23,8 +23,8 @@ min_airflow_version: "3.3"
 # JAR manifest. The render hook fails if the two disagree.
 supervisor_schema_version: "2026-10-30"
 
-# The runtime terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); it does
-# not yet emit skipped, DeferTask, RescheduleTask, or AwaitInputTask.
+# The runtime terminates a task with SucceedTask, RetryTask, DeferTask, or TaskState
+# (failed/skipped/removed); it does not yet emit RescheduleTask or AwaitInputTask.
 states:
   success:
     supported: true
@@ -37,11 +37,13 @@ states:
     since: "3.3"
     note: "RetryTask"
   skipped:
-    supported: false
-    note: "runtime does not emit TaskState skipped yet"
+    supported: true
+    since: "3.4"
+    note: "a TriggerDagRun task that skips an existing run"
   deferred:
-    supported: false
-    note: "runtime does not emit DeferTask yet"
+    supported: true
+    since: "3.4"
+    note: "a TriggerDagRun task that waits with deferrable"
   up_for_reschedule:
     supported: false
     note: "runtime does not emit RescheduleTask yet"

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

@@ -30,7 +30,8 @@ public static Bundle build() {
         .register(AnnotationExample.class)
         .register(XComCastingExample.class)
         .register(org.apache.airflow.example.nativedag.AnnotationExample.class)
-        .register(org.apache.airflow.example.nativedag.InterfaceExample.build());
+        .register(org.apache.airflow.example.nativedag.InterfaceExample.build())
+        .register(org.apache.airflow.example.nativedag.TargetExample.build());
   }
 
   public static void main(String[] args) {

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

@@ -85,6 +85,13 @@ public Class<? extends Task> pickReport(long transformed) {
         : AnnotationExampleBuilder.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")
+  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")
@@ -102,8 +109,10 @@ static class Wiring implements AnnotationExampleDeps {
     void depends() {
       var extracted = extract();
       var transformed = transform(extracted, lit(1.5));
-      hasRows(transformed).Then(load(transformed)).Else(loadEmpty());
+      var loaded = load(transformed);
+      hasRows(transformed).Then(loaded).Else(loadEmpty());
       pickReport(transformed).Case(reportLong()).Case(reportShort());
+      loaded.before(triggerDownstream());
       // Ordering-only edge: the checks group runs after extract, with no data
       // flowing.
       extracted.before(checks());

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

@@ -114,12 +114,18 @@ public static DagDef build() {
 
     var reportLong = dag.task("report_long", ReportLong.class);
     var reportShort = dag.task("report_short", ReportShort.class);
+    // A task that starts a run of another Dag; it runs no Java code.
+    var trigger =
+        dag.task(
+            "trigger_downstream",
+            new TriggerDagRun("java_native_target_example").config("wait_for_completion", false));
 
     transform.after(extract);
     // With no task id given, a decider takes one from its class: "hasRows" and
     // "pickReport".
     dag.If(HasRows.class).after(transform).Then(load).Else(loadEmpty);
     dag.Switch(PickReport.class).after(transform).Case(reportLong).Case(reportShort);
+    load.before(trigger);
     return dag;
   }
 }

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

@@ -0,0 +1,49 @@
+/*
+ * 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 java.util.List;
+import org.apache.airflow.sdk.*;
+
+// The Dag the other two native examples trigger. It triggers nothing itself,
+// so the examples cannot trigger each other in a loop.
+public class TargetExample {
+  private static final System.Logger log = System.getLogger(TargetExample.class.getName());
+
+  public static class Receive implements Task {
+    @Override
+    public void execute(Context context, Client client) {
+      log.log(INFO, "Triggered by another Dag");
+    }
+  }
+
+  public static DagDef build() {
+    var dag =
+        new DagDef("java_native_target_example")
+            .config("description", "Pure-Java Dag that the other native examples trigger")
+            .config("catchup", false)
+            .config("tags", List.of("example", "java-sdk"));
+    dag.task("receive", Receive.class);
+    return dag;
+  }
+}

java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt

@@ -278,7 +278,7 @@ class BuilderProcessor : AbstractProcessor() {
     )
     builderClass.addMethod(buildMethod.build())
 
-    declarations.forEach { builderClass.addType(buildTask(it)) }
+    declarations.filterNot { it.kind == TaskKind.TRIGGER }.forEach { builderClass.addType(buildTask(it)) }
     return builderClass.build()
   }
 
@@ -366,8 +366,11 @@ class BuilderProcessor : AbstractProcessor() {
         .methodBuilder(decl.method.simpleName.toString())
         .addModifiers(Modifier.PUBLIC, Modifier.DEFAULT)
         .returns(
-          decl.kind.refType
-            ?: ParameterizedTypeName.get(TASK_HANDLE_TYPE, TypeName.get(decl.method.returnType).boxIfPossible()),
+          when {
+            decl.kind.refType != null -> decl.kind.refType
+            decl.kind == TaskKind.TRIGGER -> ParameterizedTypeName.get(TASK_HANDLE_TYPE, VOID_TYPE)
+            else -> ParameterizedTypeName.get(TASK_HANDLE_TYPE, TypeName.get(decl.method.returnType).boxIfPossible())
+          },
         )
     decl.dataParams.forEach { method.addParameter(inType(it.type), it.name) }
     val def = taskDefCode(decl, CodeBlock.of($$"$T.$L", builderName, decl.className))
@@ -411,7 +414,13 @@ class BuilderProcessor : AbstractProcessor() {
     val taskDef =
       CodeBlock
         .builder()
-        .add($$"new $T($S, $L.class)", TASK_DEF_TYPE, decl.id, classRef)
+        .apply {
+          if (decl.kind == TaskKind.TRIGGER) {
+            add($$"new $T($S, new $T().$L())", TASK_DEF_TYPE, decl.id, ClassName.get(decl.owner), decl.method.simpleName)
+          } else {
+            add($$"new $T($S, $L.class)", TASK_DEF_TYPE, decl.id, classRef)
+          }
+        }
     explicitConfig(decl.method, decl.kind.annotation, TASK_STRUCTURAL_ATTRIBUTES, SchemaFields.TASK).forEach { (key, value) ->
       taskDef.add($$".config($S, $L)", key, value)
     }
@@ -438,17 +447,33 @@ class BuilderProcessor : AbstractProcessor() {
     for (inner in el.enclosedElements) {
       if (inner !is ExecutableElement) continue
       val declared =
-        TaskKind.entries.filter { kind -> inner.annotationMirrors.any { it.names(kind.annotation) } }
+        TaskKind.declaring.filter { kind -> inner.annotationMirrors.any { it.names(kind.annotation) } }
       if (declared.isEmpty()) continue
-      val kind =
+      val declaredKind =
         declared.singleOrNull()
           ?: throw IllegalArgumentException(
             "Method '${inner.simpleName}' carries ${declared.joinToString { it.spelling }}; a task is " +
               "declared by one of them alone",
           )
+      // A task method that hands back a TriggerDagRun declares what to trigger
+      // rather than a body to run, so it is read when the Dag is built.
+      val triggers =
+        declaredKind == TaskKind.TASK && with(processingEnv) { isType(inner.returnType, TRIGGER_TYPE) }
+      val kind = if (triggers) TaskKind.TRIGGER else declaredKind
       val annotated = declaredId(inner, kind)
       if (inner.isVarArgs) throw IllegalArgumentException("Cannot create task from vararg function ${inner.simpleName}")
       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 " +
+            "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 " +
+            "is built, where a checked exception cannot be thrown; it must not throw: ${checked.joinToString()}"
+        }
+      }
       val localId = annotated.ifBlank { inner.simpleName.toString() }
       require(tasks.none { it.method.simpleName.contentEquals(inner.simpleName) }) {
         "Class ${el.simpleName} overloads task method '${inner.simpleName}'; a method's name is the " +
@@ -572,7 +597,8 @@ class BuilderProcessor : AbstractProcessor() {
       }
     }
     val byClassName = mutableMapOf<String, TaskDeclaration>()
-    declarations.forEach { decl ->
+    // A task that triggers a Dag run has no generated class to collide with.
+    declarations.filterNot { it.kind == TaskKind.TRIGGER }.forEach { decl ->
       byClassName.put(decl.className, decl)?.let { first ->
         throw IllegalArgumentException(
           "Task methods '${first.id}' and '${decl.id}' both generate the task class " +
@@ -694,6 +720,8 @@ class BuilderProcessor : AbstractProcessor() {
         parseTemporal(field, text) { OffsetDateTime.parse(text) }
         CodeBlock.of($$"$T.parse($S)", ClassName.get(OffsetDateTime::class.java), text)
       }
+      FieldType.DAG_RUN_STATES, FieldType.JSON_OBJECT ->
+        error("Field '${field.key}' is not an annotation attribute")
     }
 
   private fun parseTemporal(
@@ -718,7 +746,7 @@ class BuilderProcessor : AbstractProcessor() {
   ) {
     val returns = method.returnType
     when (kind) {
-      TaskKind.TASK -> return
+      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: " +
@@ -760,7 +788,7 @@ class BuilderProcessor : AbstractProcessor() {
         .addModifiers(Modifier.PUBLIC)
         .returns(
           when (decl.kind) {
-            TaskKind.TASK -> TypeName.VOID
+            TaskKind.TASK, TaskKind.TRIGGER -> TypeName.VOID
             TaskKind.CONDITION -> TypeName.BOOLEAN
             TaskKind.SWITCH -> TASK_CLASS_TYPE
           },
@@ -1008,6 +1036,8 @@ private val SWITCH_TASK_TYPE = ClassName.get(SwitchTask::class.java)
 private val SWITCH_REF_TYPE = ClassName.get(SwitchRef::class.java)
 private val TASK_CLASS_TYPE =
   ParameterizedTypeName.get(ClassName.get(Class::class.java), WildcardTypeName.subtypeOf(TASK_TYPE))
+private val TRIGGER_TYPE = ClassName.get(TriggerDagRun::class.java)
+private val VOID_TYPE = ClassName.get("java.lang", "Void")
 private val BOXED_BOOLEAN_TYPE = ClassName.get("java.lang", "Boolean")
 private val DEPS_TYPE = ClassName.get(Deps::class.java)
 private val GROUP_TYPE = DEPS_TYPE.nestedClass("TaskGroup")
@@ -1029,8 +1059,18 @@ private enum class TaskKind(
   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),
+  ;
+
+  companion object {
+    /**
+     * The kinds an annotation declares. TRIGGER is left out: it shares
+     * [TASK]'s annotation and is told apart by the method's return type.
+     */
+    val declaring = listOf(TASK, CONDITION, SWITCH)
+  }
 }
 
 /** Whether this annotation is the one [name] qualifies. */