Java SDK: Let a native Java Dag set one queue for all its tasks (#74538)

* Java SDK: Let a native Java Dag set one queue for all its tasks

Every task of a native Java Dag runs on the Java coordinator, so each one
needs a queue that queue_to_coordinator routes there. Airflow's Dag has no
queue setting, so authors had to repeat the same queue on every task. The
Go and TypeScript SDKs already take a Dag-level queue that each task
inherits unless it sets its own; this gives Java the same.

* Reject a blank Dag-level queue and fix the native Dag queue docs

A blank queue matched no coordinator, so every task stayed queued.
The parsing docs still said there was no Dag-level queue and routed a
queue the examples do not use.

* Reject a blank Dag-level queue in the annotation processor

* Carry the blank rule on the schema field, for Dag and task queues alike

* Set the queue once per Dag in the e2e native bundle

* Put the blank check on Field and tidy the comments

---------

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

@@ -31,6 +31,7 @@ import org.apache.airflow.sdk.*;
     id = "java_native_annotation_example",
     description = "Pure-Java Dag authored with annotations",
     schedule = "@daily",
+    queue = "java",
     startDate = "2026-01-01T00:00:00Z",
     catchup = false,
     tags = {"example", "java-sdk"})

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

@@ -101,6 +101,7 @@ public class InterfaceExample {
         new DagDef("java_native_interface_example")
             .config("description", "Pure-Java Dag authored with the interface API")
             .config("schedule", "@daily")
+            .config("queue", "java")
             .config("catchup", false)
             .config("tags", List.of("example", "java-sdk"));
 

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

@@ -41,6 +41,7 @@ public class TargetExample {
     var dag =
         new DagDef("java_native_target_example")
             .config("description", "Pure-Java Dag that the other native examples trigger")
+            .config("queue", "java")
             .config("catchup", false)
             .config("tags", List.of("example", "java-sdk"));
     dag.task("receive", Receive.class);

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

@@ -712,7 +712,11 @@ class AirflowProcessor : AbstractProcessor() {
     value: AnnotationValue,
   ): CodeBlock =
     when (field.type) {
-      FieldType.STRING -> CodeBlock.of($$"$S", value.value)
+      FieldType.STRING -> {
+        val text = value.value as String
+        field.checkString(text) { "Annotation attribute '${field.attribute}'" }
+        CodeBlock.of($$"$S", text)
+      }
       FieldType.BOOLEAN, FieldType.INTEGER, FieldType.NUMBER -> CodeBlock.of($$"$L", value.value)
       FieldType.STRING_ARRAY -> {
         @Suppress("UNCHECKED_CAST")

java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt

@@ -403,7 +403,7 @@ class AirflowProcessorTest {
         """
         package org.apache.airflow.example;
         import org.apache.airflow.sdk.Airflow;
-        @Airflow.Dag(id = "cfg", schedule = "@daily", tags = {"a", "b"}, catchup = true,
+        @Airflow.Dag(id = "cfg", schedule = "@daily", queue = "java", tags = {"a", "b"}, catchup = true,
             startDate = "2026-01-01T00:00:00Z")
         public class TestExample {
           @Airflow.Task(retries = 2, queue = "q", retryDelay = "PT5M", retryExponentialBackoff = 1.5)
@@ -442,6 +442,7 @@ class AirflowProcessorTest {
            public static DagDef build() {
              var dag = DagSource.declaredBy(new DagDef("cfg"), TestExample.class);
              dag.config("schedule", "@daily");
+             dag.config("queue", "java");
              dag.config("tags", List.of("a", "b"));
              dag.config("catchup", true);
              dag.config("start_date", OffsetDateTime.parse("2026-01-01T00:00:00Z"));
@@ -2074,6 +2075,56 @@ class AirflowProcessorTest {
     )
   }
 
+  @Test
+  @DisplayName("reject a blank Dag-level queue")
+  fun rejectBlankDagQueue() {
+    for (queue in listOf("", " ")) {
+      val compilation =
+        compile(
+          """
+          package org.apache.airflow.example;
+          import org.apache.airflow.sdk.Airflow;
+          @Airflow.Dag(queue = "$queue")
+          public class TestExample {
+            @Airflow.Task public void t() {}
+
+            @Airflow.Deps
+            static class Wiring implements TestExampleDeps {
+              void depends() { t(); }
+            }
+          }
+        """,
+        )
+      assertThat(compilation).failed()
+      assertThat(compilation).hadErrorContaining("Annotation attribute 'queue' must not be blank")
+    }
+  }
+
+  @Test
+  @DisplayName("reject a blank task-level queue")
+  fun rejectBlankTaskQueue() {
+    for (queue in listOf("", " ")) {
+      val compilation =
+        compile(
+          """
+          package org.apache.airflow.example;
+          import org.apache.airflow.sdk.Airflow;
+          @Airflow.Dag
+          public class TestExample {
+            @Airflow.Task(queue = "$queue") public void t() {}
+
+            @Airflow.Deps
+            static class Wiring implements TestExampleDeps {
+              void depends() { t(); }
+            }
+          }
+        """,
+        )
+      assertThat(compilation).failed()
+      assertThat(compilation).hadErrorContaining("Annotation attribute 'queue' must not be blank")
+    }
+  }
+
   @Test
   @DisplayName("escape quotes and backslashes in a string-array attribute")
   fun generateBuilderEscapesStringArrayValues() {

java-sdk/sdk/build.gradle.kts

@@ -284,8 +284,15 @@ abstract class GenerateDagDslTask : DefaultTask() {
         val attrDefault: String,
         val defaultJson: String?,
         val doc: String,
+        val notBlank: Boolean = false,
     )
 
+    // Schema keys whose value cannot be blank, in every table that holds them.
+    private val notBlankKeys = setOf("queue")
+
+    private fun constrained(fields: List<DslField>): List<DslField> =
+        fields.map { if (it.key in notBlankKeys) it.copy(notBlank = true) else it }
+
     private fun camelCase(key: String): String =
         key
             .split('_')
@@ -461,6 +468,22 @@ abstract class GenerateDagDslTask : DefaultTask() {
                         "`\"@once\"`, `\"@continuous\"`, a cron expression, or empty for no schedule.",
                     ),
                 )
+                // The schema has no Dag-level queue; a Python Dag sets one through default_args instead.
+                // Every task of a Java Dag runs on the Java coordinator, so one queue on the Dag routes all of them there.
+                if (!dagProps.path("queue").isMissingNode) {
+                    throw GradleException("The schema now has a Dag-level 'queue'; resolve it from the schema instead")
+                }
+                add(
+                    DslField(
+                        "queue",
+                        "queue",
+                        "STRING",
+                        "String",
+                        quote(""),
+                        null,
+                        "Queue each task of the Dag runs on, unless the task sets its own `queue`. It cannot be blank.",
+                    ),
+                )
                 dagAllowlist.forEach { key ->
                     val prop = dagProps.path(key)
                     if (prop.isMissingNode) {
@@ -471,7 +494,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
                             ?: throw GradleException("Dag allowlist key '$key' is not a scalar the DSL can express"),
                     )
                 }
-            }
+            }.let(::constrained)
 
         val excludedSeen = mutableSetOf<String>()
         val taskFields =
@@ -490,7 +513,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
                     val override = if (key == "retry_exponential_backoff") "Double" else null
                     resolveField(key, prop, override)?.let { add(it) }
                 }
-            }
+            }.let(::constrained)
         (excludedTaskKeys - excludedSeen).takeIf { it.isNotEmpty() }?.let {
             throw GradleException("Excluded task keys match no eligible schema property; remove or fix: $it")
         }
@@ -756,6 +779,7 @@ abstract class GenerateDagDslTask : DefaultTask() {
                     "          \"${f.attribute}\",",
                     "          FieldType.${f.fieldType},",
                     "          $defaultRepr,",
+                    "          ${f.notBlank},",
                     "        ),",
                 ).joinToString("\n")
             }

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

@@ -87,13 +87,16 @@ class DagDef(
    * mismatched value types are rejected on the call, so mistakes surface where
    * the Dag is defined.
    *
+   * `"queue"` is the one key Airflow's Dag has no setting for.
+   * It is the queue every task of the Dag runs on, unless the task sets its own `"queue"`, and it cannot be blank.
+   *
    * @param key Airflow Dag setting name.
    * @param value Value matching the key's schema type. Durations take
    *    [java.time.Duration], date-times [java.time.OffsetDateTime] or
    *    [java.time.Instant], string arrays any `Iterable` of `String`.
    * @return This Dag, for chaining.
-   * @throws IllegalArgumentException if the key is unknown or the value type
-   *    does not match.
+   * @throws IllegalArgumentException if the key is unknown, the value type
+   *    does not match, or `"queue"` is blank.
    */
   fun config(
     key: String,
@@ -542,17 +545,17 @@ class TaskDef(
    * Sets one task-level configuration value.
    *
    * Keys are Airflow's own task setting names (for example `"retries"`,
-   * `"queue"`, `"retry_delay"`); unknown keys and mismatched
-   * value types are rejected on the call, so mistakes surface where the task is
-   * defined.
+   * `"queue"`, `"retry_delay"`); unknown keys, mismatched
+   * value types and a blank `"queue"` are rejected on the call, so mistakes
+   * surface where the task is defined.
    *
    * @param key Airflow task setting name, e.g. `"retries"`.
    * @param value Value matching the key's schema type. Durations take
    *    [java.time.Duration], date-times [java.time.OffsetDateTime] or
    *    [java.time.Instant].
    * @return This task definition, for chaining.
-   * @throws IllegalArgumentException if the key is unknown or the value type
-   *    does not match.
+   * @throws IllegalArgumentException if the key is unknown, the value type
+   *    does not match, or `"queue"` is blank.
    */
   fun config(
     key: String,

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

@@ -66,6 +66,7 @@ private val TRIGGER_FIELDS: Map<String, Field> =
  * The task runs no Java code and takes no arguments. It pushes the triggered
  * run's ID, and the link the "Triggered DAG" extra link reads. It renders no
  * templates, so a value such as `"{{ ds }}"` reaches the new run unchanged.
+ * Like a Java task, it runs on the Dag's `queue` unless the task sets its own.
  *
  * @param dagId `trigger_dag_id`: the Dag to trigger.
  * @throws IllegalArgumentException if [dagId] is empty.

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt

@@ -102,6 +102,7 @@ internal fun serializeDag(
     }
   }
 
+  val dagQueue = dag.dagConfig["queue"] as String?
   val result =
     linkedMapOf<String, Any?>(
       "dag_id" to dag.id,
@@ -109,7 +110,7 @@ internal fun serializeDag(
       "relative_fileloc" to relativeFileloc,
       "timezone" to dagTimezone(dag.dagConfig),
       "timetable" to serializeTimetable(dag.id, dag.dagConfig),
-      "tasks" to dag.tasks.map { (taskId, def) -> serializeTask(taskId, def, downstream[taskId]) },
+      "tasks" to dag.tasks.map { (taskId, def) -> serializeTask(taskId, def, downstream[taskId], dagQueue) },
       "dag_dependencies" to serializeDagDependencies(dag),
       "task_group" to serializeTaskGroups(dag, expansion),
       "edge_info" to serializeEdgeInfo(dag, expansion),
@@ -124,11 +125,13 @@ internal fun serializeDag(
 /**
  * Converts one task to the Airflow serialization format. `downstream` is the
  * inverted view of the Dag's upstream edges, sorted for stable JSON.
+ * `dagQueue` is the Dag's queue, which the task takes unless it sets its own.
  */
 private fun serializeTask(
   taskId: String,
   def: TaskDef,
   downstream: List<String>?,
+  dagQueue: String?,
 ): Map<String, Any?> {
   val data = linkedMapOf<String, Any?>("task_id" to taskId)
   val trigger = def.trigger
@@ -150,13 +153,16 @@ private fun serializeTask(
   }
   // Lets NotPreviouslySkippedDep re-skip a cleared downstream, as Python's SkipMixin does.
   if (def.decider != null) data["_can_skip_downstream"] = true
+  // A trigger task takes the Dag's queue too, because the Java runtime runs it.
+  val config =
+    if (dagQueue == null || "queue" in def.configValues) def.configValues else def.configValues + ("queue" to dagQueue)
   // Emit only config entries that differ from their schema default, mirroring
   // Python BaseSerialization's "omit hard-coded default" behavior, which the Go
   // and TypeScript SDKs mirror too. Operator fields are stored unwrapped, so the
   // __type encoding is stripped. If core grows a task-level fill_config_defaults,
   // every SDK has to keep explicitly set values instead, or an explicit retries=0
   // reads as unset and picks up the configured default.
-  def.configValues.forEach { (key, value) ->
+  config.forEach { (key, value) ->
     if (key !in OMITTED_TASK_KEYS && !matchesSchemaDefault(SchemaFields.TASK[key], value)) {
       data[key] = unwrapTypeEncoding(serializeValue(value))
     }

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt

@@ -57,13 +57,28 @@ private val DAG_RUN_STATE_NAMES = listOf("queued", "running", "success", "failed
  * @property type Accepted value shape.
  * @property defaultJson Schema default as raw JSON, or `null` when the schema
  *    declares no default.
+ * @property notBlank Whether a value for this key must not be blank.
  */
 class Field(
   val key: String,
   val attribute: String,
   val type: FieldType,
   val defaultJson: String?,
-)
+  val notBlank: Boolean = false,
+) {
+  /**
+   * Enforces the constraints this key carries on a string value. [describe]
+   * names the value, so each caller keeps its own wording.
+   *
+   * @throws IllegalArgumentException if [text] breaks a constraint.
+   */
+  fun checkString(
+    text: String,
+    describe: () -> String,
+  ) {
+    require(!notBlank || text.isNotBlank()) { "${describe()} must not be blank" }
+  }
+}
 
 /**
  * Validates one `config(key, value)` call against a schema field table and
@@ -91,7 +106,10 @@ internal fun checkConfigValue(
       "Value for $scope config key '$key' must be $expected, got: ${value.javaClass.name}",
     )
   return when (field.type) {
-    FieldType.STRING -> value as? String ?: mismatch("a String")
+    FieldType.STRING ->
+      (value as? String ?: mismatch("a String")).also { text ->
+        field.checkString(text) { "Value for $scope config key '$key'" }
+      }
     FieldType.BOOLEAN -> value as? Boolean ?: mismatch("a Boolean")
     FieldType.NUMBER -> value as? Number ?: mismatch("a Number")
     FieldType.INTEGER ->

java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/DagDefTest.kt

@@ -175,6 +175,7 @@ internal class DagDefTest {
         .config("dagrun_timeout", Duration.ofMinutes(5))
         .config("start_date", OffsetDateTime.parse("2026-01-01T00:00:00Z"))
         .config("tags", listOf("a", "b"))
+        .config("queue", "java")
 
     Assertions.assertEquals(
       mapOf(
@@ -185,6 +186,7 @@ internal class DagDefTest {
         "dagrun_timeout" to Duration.ofMinutes(5),
         "start_date" to OffsetDateTime.parse("2026-01-01T00:00:00Z"),
         "tags" to listOf("a", "b"),
+        "queue" to "java",
       ),
       dag.dagConfig,
     )
@@ -226,6 +228,32 @@ internal class DagDefTest {
     Assertions.assertEquals("Value for Dag config key 'description' must not be null", error.message)
   }
 
+  @Test
+  @DisplayName("Should reject a blank dag queue")
+  fun shouldRejectBlankDagQueue() {
+    for (queue in listOf("", "  ")) {
+      val error =
+        Assertions.assertThrows(IllegalArgumentException::class.java) {
+          DagDef("dag").config("queue", queue)
+        }
+
+      Assertions.assertEquals("Value for Dag config key 'queue' must not be blank", error.message)
+    }
+  }
+
+  @Test
+  @DisplayName("Should reject a blank task queue")
+  fun shouldRejectBlankTaskQueue() {
+    for (queue in listOf("", "  ")) {
+      val error =
+        Assertions.assertThrows(IllegalArgumentException::class.java) {
+          DagDef("dag").task<Unit>("extract", NoOp::class.java).config("queue", queue)
+        }
+
+      Assertions.assertEquals("Value for task config key 'queue' must not be blank", error.message)
+    }
+  }
+
   @Test
   @DisplayName("Should reject non-integral values for integer dag config keys")
   fun shouldRejectFractionalIntegerValue() {

java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt

@@ -22,10 +22,13 @@ package org.apache.airflow.sdk.execution
 import org.apache.airflow.sdk.Arg
 import org.apache.airflow.sdk.Bundle
 import org.apache.airflow.sdk.Client
+import org.apache.airflow.sdk.ConditionTask
 import org.apache.airflow.sdk.Context
 import org.apache.airflow.sdk.DagDef
+import org.apache.airflow.sdk.SwitchTask
 import org.apache.airflow.sdk.Task
 import org.apache.airflow.sdk.TaskDef
+import org.apache.airflow.sdk.TriggerDagRun
 import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
 import org.apache.airflow.sdk.internal.Refs
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -44,6 +47,20 @@ private class SerdeNoopTask : Task {
   ) = Unit
 }
 
+private class SerdeCondition : ConditionTask {
+  override fun decide(
+    context: Context,
+    client: Client,
+  ) = true
+}
+
+private class SerdeSwitch : SwitchTask {
+  override fun choose(
+    context: Context,
+    client: Client,
+  ) = SerdeNoopTask::class.java
+}
+
 @Suppress("UNCHECKED_CAST")
 private fun taskData(
   serialized: Map<String, Any?>,
@@ -54,6 +71,12 @@ private fun taskData(
   return tasks[index]["__var"] as Map<String, Any?>
 }
 
+@Suppress("UNCHECKED_CAST")
+private fun queuesByTaskId(serialized: Map<String, Any?>): Map<String, Any?> =
+  (serialized["tasks"] as List<Map<String, Any?>>)
+    .map { it["__var"] as Map<String, Any?> }
+    .associate { it["task_id"] as String to it["queue"] }
+
 internal class SerdeTest {
   @Test
   @DisplayName("Should emit required dag fields and leave unset config-backed fields out")
@@ -186,6 +209,47 @@ internal class SerdeTest {
     )
   }
 
+  @Test
+  @DisplayName("Should give every task the Dag's queue unless the task sets its own")
+  fun shouldGiveEachTaskTheDagQueue() {
+    val dag = DagDef("d").config("queue", "java")
+    val extract = dag.task<Unit>("extract", SerdeNoopTask::class.java)
+    dag.task<Unit>("heavy", SerdeNoopTask::class.java).config("queue", "java_large")
+    dag.task<Unit>("on_default", SerdeNoopTask::class.java).config("queue", "default")
+    dag.If("has_rows", SerdeCondition::class.java).after(extract)
+    dag.Switch("pick", SerdeSwitch::class.java).after(extract)
+    dag.task("trigger", TriggerDagRun("reports"))
+    dag.task("trigger_on_python", TriggerDagRun("reports")).config("queue", "python")
+
+    val serialized = serializeDag(dag, "", ".")
+
+    assertEquals(
+      mapOf(
+        "extract" to "java",
+        "heavy" to "java_large",
+        "on_default" to null,
+        "has_rows" to "java",
+        "pick" to "java",
+        "trigger" to "java",
+        "trigger_on_python" to "python",
+      ),
+      queuesByTaskId(serialized),
+    )
+    assertFalse("queue" in serialized)
+  }
+
+  @Test
+  @DisplayName("Should leave the queue out when the Dag's queue is the schema default or unset")
+  fun shouldLeaveOutDefaultOrUnsetDagQueue() {
+    val onDefault = DagDef("d").config("queue", "default")
+    onDefault.task<Unit>("t", SerdeNoopTask::class.java)
+    val unset = DagDef("d")
+    unset.task<Unit>("t", SerdeNoopTask::class.java)
+
+    assertFalse("queue" in taskData(serializeDag(onDefault, "", "."), 0))
+    assertFalse("queue" in taskData(serializeDag(unset, "", "."), 0))
+  }
+
   @Test
   @DisplayName("Should serialize nested task groups with their own edges")
   fun shouldSerializeTaskGroups() {