Java SDK: Generate Dag and task configuration from the serialization schema

.pre-commit-config.yaml

@@ -337,6 +337,17 @@ repos:
           (?x)
           ^airflow-core/src/airflow/serialization/schema\.json$|
           ^ts-sdk/schema/dag-schema\.json$
+      - id: sync-java-sdk-dag-schema
+        name: Sync Java SDK Dag serialization schema with airflow-core
+        description: "Copy airflow-core's serialization schema when Java SDK's vendored dag-schema.json drifts"
+        entry: ./java-sdk/gradlew -p ./java-sdk :sdk:syncDagSchema
+        language: system
+        pass_filenames: false
+        # Re-vendoring is a java-sdk maintainer's deliberate step after an airflow-core
+        # schema change, not something every commit should do, so this is manual only:
+        #   prek run sync-java-sdk-dag-schema --hook-stage manual
+        stages: ['manual']
+        files: ^java-sdk/sdk/schema/dag-schema\.json$
       - id: check-go-version-in-sync
         name: Check Go toolchain version is consistent across build files
         entry: ./scripts/ci/prek/check_go_version_in_sync.py

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

@@ -306,6 +306,14 @@ Annotate a plain Java class and let the SDK generate the boilerplate at compile
        ``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 = "...")``
+     - 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 = "...")``
+     - 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.
    * - ``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
@@ -348,7 +356,8 @@ Interface-based API
 ~~~~~~~~~~~~~~~~~~~
 
 Implement the ``Task`` interface directly for full control over how tasks are registered and how XComs are
-read.  Each task is registered as a ``TaskDef`` on a ``DagDef``.
+read.  Each task is registered as a ``TaskDef`` on a ``DagDef``; both carry a fluent
+``config(key, value)`` whose keys are Airflow's own setting names.
 
 The runner creates a fresh instance of the task class through reflection for every task-instance run,
 which puts four constraints on the class:
@@ -537,16 +546,15 @@ calls with no arguments.
 Native Java Dags
 ----------------
 
-A Dag can also be authored entirely in Java, with no Python stub file: the ``DagDef`` and
-``TaskDef`` objects hold the tasks, and Java declares the graph.
+A Dag can also be authored entirely in Java: the annotations (or the ``DagDef`` / ``TaskDef``
+objects) carry the configuration, and Java declares the graph.
 
 Building the Dag in Java
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
-``dag.task(...)`` registers a task as it creates it and hands back a handle, so there is no second
-``addTask`` call to forget.  ``before`` and ``after`` draw every edge on this surface — Python's
-``a >> b`` and ``b << a`` — and the task body moves the data itself, by reading the upstream's XCom
-through ``Client``:
+``dag.task(...)`` registers a task as it creates it and hands back a handle.  ``before`` and
+``after`` draw every edge on this surface, and the task body moves the data itself, by reading the
+upstream's XCom through ``Client``:
 
 .. code-block:: java
 
@@ -566,6 +574,21 @@ 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.
 
+Configuration attributes
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+The ``@Builder.Dag`` and ``@Builder.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"``).
+Only attributes written explicitly at the use site are applied, so Airflow's own defaults still
+apply to everything left out.
+
+Durations and date-times are ISO-8601 strings in annotations (``retryDelay = "PT5M"``,
+``startDate = "2026-01-01T00:00:00Z"``, validated at compile time) and ``java.time.Duration`` /
+``java.time.OffsetDateTime`` values in ``config`` calls.  An unknown key or a mismatched value type
+fails the build for an annotation, and the ``config`` call itself for an object.
+
 .. _java-sdk/logging:
 
 Logging

java-sdk/README.md

@@ -726,9 +726,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`. When adding a
-  new annotation, define it in `Builder.kt`, handle it in
-  `BuilderProcessor.kt`, and add a golden-output test in
+- 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
+  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
   the JVM process from Python beyond what `_build_execute_task_command`
@@ -750,9 +754,14 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
 5. Update `airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst`
    if the change is user-visible.
 
-**Adding a new annotation**:
+**Adding a new annotation or configuration attribute**:
 
-1. Define the annotation interface in `Builder.kt`.
+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
+   `: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
    `*Builder` class.
 3. Add a test in `BuilderTest.kt` with expected generated output.

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

@@ -22,11 +22,11 @@
 
 import static java.lang.System.Logger.Level.INFO;
 
+import java.util.List;
 import org.apache.airflow.sdk.*;
 
-// A Dag defined entirely in Java, interface-style: no Python stub file
-// describes it. dag.task registers a task as it creates it and hands back the
-// handle, and `before`/`after` wire the graph -- Java's spelling of `>>` and `<<`.
+// A Dag defined entirely in Java, interface-style. dag.task registers a task as
+// it creates it and hands back the handle, and `before`/`after` wire the graph.
 public class InterfaceExample {
   private static final System.Logger log = System.getLogger(InterfaceExample.class.getName());
 
@@ -56,9 +56,17 @@ public void execute(Context context, Client client) {
   }
 
   public static DagDef build() {
-    var dag = new DagDef("java_native_interface_example");
+    var dag =
+        new DagDef("java_native_interface_example")
+            .config("description", "Pure-Java Dag authored with the interface API")
+            .config("schedule", "@daily")
+            .config("catchup", false)
+            .config("tags", List.of("example", "java-sdk"));
 
-    var extract = dag.task("extract", Extract.class);
+    var extract =
+        dag.task("extract", Extract.class)
+            .config("retries", 2)
+            .config("doc_md", "Extracts a value and pushes it as an XCom.");
     var transform = dag.task("transform", Transform.class);
     var load = dag.task("load", Load.class);
 

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

@@ -29,16 +29,24 @@ import com.squareup.javapoet.ParameterizedTypeName
 import com.squareup.javapoet.TypeName
 import com.squareup.javapoet.TypeSpec
 import org.apache.airflow.sdk.internal.ArgValues
+import org.apache.airflow.sdk.internal.Field
+import org.apache.airflow.sdk.internal.FieldType
+import org.apache.airflow.sdk.internal.SchemaFields
 import org.apache.airflow.sdk.internal.TaskArgs
 import org.apache.airflow.sdk.internal.TypeRef
 import org.apache.airflow.sdk.internal.foldArgName
 import org.apache.airflow.sdk.internal.registrarName
+import java.time.Duration
+import java.time.OffsetDateTime
+import java.time.format.DateTimeParseException
 import javax.annotation.processing.AbstractProcessor
 import javax.annotation.processing.ProcessingEnvironment
 import javax.annotation.processing.RoundEnvironment
 import javax.annotation.processing.SupportedAnnotationTypes
 import javax.annotation.processing.SupportedSourceVersion
 import javax.lang.model.SourceVersion
+import javax.lang.model.element.AnnotationValue
+import javax.lang.model.element.Element
 import javax.lang.model.element.ElementKind
 import javax.lang.model.element.ExecutableElement
 import javax.lang.model.element.Modifier
@@ -61,13 +69,15 @@ import javax.tools.Diagnostic
  * containing:
  *
  * - One inner class per [Builder.Task]-annotated method, implementing [Task].
- * - A static `build()` method that constructs the [DagDef] and registers those
- *   inner classes as [TaskDef]s.
+ * - A static `build()` method that constructs the [DagDef], lowers every
+ *   explicitly-written `@Builder.Dag` attribute into a `DagDef.config` call,
+ *   and registers those inner classes as [TaskDef]s, each carrying its
+ *   explicitly-written `@Builder.Task` attributes the same way.
  *
  * In the generated `execute` body, a task's data parameters resolve against the
  * arg bindings the supervisor delivered for the run: flat parameters through
  * [TaskArgs], by their position among the data parameters, and [TaskInput]
- * [TaskInput] fields through [ArgValues], by argument name. Non-`void` return values are
+ * fields through [ArgValues], by argument name. Non-`void` return values are
  * forwarded to `client.setXCom`.
  */
 @SupportedAnnotationTypes(
@@ -177,6 +187,9 @@ class BuilderProcessor : AbstractProcessor() {
         .addModifiers(Modifier.PUBLIC, Modifier.STATIC)
         .returns(DAG_DEF_TYPE)
         .addStatement($$"var dag = new $T($S)", DAG_DEF_TYPE, ann.id.ifBlank { el.simpleName })
+    explicitConfig(el, DAG_ANNOTATION, DAG_STRUCTURAL_ATTRIBUTES, SchemaFields.DAG).forEach { (key, value) ->
+      buildMethod.addStatement($$"dag.config($S, $L)", key, value)
+    }
 
     for (inner in el.enclosedElements) {
       if (inner !is ExecutableElement) continue
@@ -188,10 +201,8 @@ class BuilderProcessor : AbstractProcessor() {
       builderClass.addType(buildTask(innerName, inner, el))
 
       buildMethod.addStatement(
-        $$"dag.addTask(new $T($S, $L.class))",
-        TASK_DEF_TYPE,
-        taskAnn.id.ifBlank { inner.simpleName },
-        innerName,
+        $$"dag.addTask($L)",
+        taskDefCode(inner, taskAnn.id.ifBlank { inner.simpleName.toString() }, innerName),
       )
     }
 
@@ -200,6 +211,93 @@ class BuilderProcessor : AbstractProcessor() {
     return builderClass.build()
   }
 
+  /**
+   * Emits `new TaskDef(id, <className>.class)` with the explicitly-written
+   * `@Builder.Task` attributes lowered into chained `.config` calls.
+   */
+  private fun taskDefCode(
+    method: ExecutableElement,
+    id: String,
+    className: String,
+  ): CodeBlock {
+    val taskDef =
+      CodeBlock
+        .builder()
+        .add($$"new $T($S, $L.class)", TASK_DEF_TYPE, id, className)
+    explicitConfig(method, TASK_ANNOTATION, TASK_STRUCTURAL_ATTRIBUTES, SchemaFields.TASK).forEach { (key, value) ->
+      taskDef.add($$".config($S, $L)", key, value)
+    }
+    return taskDef.build()
+  }
+
+  /**
+   * Lowers the explicitly-written configuration attributes of [element]'s
+   * [annotationName] annotation into (schema key, value code) pairs. Only
+   * attributes present at the use site are lowered, so annotation defaults
+   * never override the schema's own defaults.
+   */
+  private fun explicitConfig(
+    element: Element,
+    annotationName: String,
+    structural: Set<String>,
+    table: Map<String, Field>,
+  ): List<Pair<String, CodeBlock>> {
+    val mirror =
+      element.annotationMirrors.firstOrNull {
+        (it.annotationType.asElement() as TypeElement).qualifiedName.contentEquals(annotationName)
+      } ?: return emptyList()
+    val byAttribute = table.values.associateBy { it.attribute }
+    return mirror.elementValues.mapNotNull { (attr, value) ->
+      val name = attr.simpleName.toString()
+      if (name in structural) return@mapNotNull null
+      val field =
+        requireNotNull(byAttribute[name]) {
+          "Annotation attribute '$name' has no Dag serialization schema key"
+        }
+      field.key to configValueCode(field, value)
+    }
+  }
+
+  private fun configValueCode(
+    field: Field,
+    value: AnnotationValue,
+  ): CodeBlock =
+    when (field.type) {
+      FieldType.STRING -> CodeBlock.of($$"$S", value.value)
+      FieldType.BOOLEAN, FieldType.INTEGER, FieldType.NUMBER -> CodeBlock.of($$"$L", value.value)
+      FieldType.STRING_ARRAY -> {
+        @Suppress("UNCHECKED_CAST")
+        val items = value.value as List<AnnotationValue>
+        CodeBlock.of(
+          $$"$T.of($L)",
+          ClassName.get(List::class.java),
+          CodeBlock.join(items.map { CodeBlock.of($$"$S", it.value) }, ", "),
+        )
+      }
+      FieldType.TIMEDELTA -> {
+        val text = value.value as String
+        parseTemporal(field, text) { Duration.parse(text) }
+        CodeBlock.of($$"$T.parse($S)", ClassName.get(Duration::class.java), text)
+      }
+      FieldType.DATETIME -> {
+        val text = value.value as String
+        parseTemporal(field, text) { OffsetDateTime.parse(text) }
+        CodeBlock.of($$"$T.parse($S)", ClassName.get(OffsetDateTime::class.java), text)
+      }
+    }
+
+  private fun parseTemporal(
+    field: Field,
+    text: String,
+    parse: () -> Any,
+  ) {
+    try {
+      parse()
+    } catch (e: DateTimeParseException) {
+      throw IllegalArgumentException("Annotation attribute '${field.attribute}' is not valid ISO-8601: '$text'")
+    }
+  }
+
   private fun buildTask(
     name: String,
     inner: ExecutableElement,
@@ -404,6 +502,12 @@ private val TASK_ARGS_TYPE = ClassName.get(TaskArgs::class.java)
 private val TYPE_REF_TYPE = ClassName.get(TypeRef::class.java)
 private val ARG_VALUES_TYPE = ClassName.get(ArgValues::class.java)
 
+private const val DAG_ANNOTATION = "org.apache.airflow.sdk.Builder.Dag"
+private const val TASK_ANNOTATION = "org.apache.airflow.sdk.Builder.Task"
+
+private val DAG_STRUCTURAL_ATTRIBUTES = setOf("id", "to")
+private val TASK_STRUCTURAL_ATTRIBUTES = setOf("id")
+
 private fun ProcessingEnvironment.isType(
   t: TypeMirror,
   c: ClassName,