jason810496 · Oct 23, 2026 · Oct 22, 2026 · Oct 22, 2026
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -352,7 +352,8 @@ Annotate a plain Java class and let the SDK generate the boilerplate at compile
    * - ``@Airflow.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`.
+       explicitly are applied.  ``queue`` is the queue each task runs on unless the task sets its
+       own.  See :ref:`java-sdk/native-dags`.
    * - ``@Airflow.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
@@ -642,6 +643,11 @@ Native Java Dags
 A Dag can also be authored entirely in Java: the annotations (or the ``DagDef`` / ``TaskDef``
 objects) carry the configuration, and Java declares the graph.
 
+Every task of the Dag runs on the Java coordinator, so it needs a queue that
+:ref:`queue_to_coordinator <java-sdk/coordinator-config>` sends there. Set ``queue`` once on the Dag, with
+``@Airflow.Dag(queue = "java")`` or ``dag.config("queue", "java")``, and each task inherits it, including a
+``TriggerDagRun`` task. A task's own ``queue`` wins over the Dag's. A ``queue`` that is set cannot be blank.
+
 Building the Dag in Java
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
@@ -922,7 +928,7 @@ The Dag processor runs the JAR's main class to list its Dags, so it needs a Java
         "kwargs": {"java_executable": "/usr/lib/jvm/java-17-openjdk/bin/java"}
       }
     }
-    queue_to_coordinator = {"java-native": "java-native"}
+    queue_to_coordinator = {"java": "java-native"}
 
 Once a ``JavaCoordinator`` is configured, the Dag processor parses the executable JARs of every Dag bundle,
 so it needs this ``[sdk]`` configuration and a JRE. With one ``JavaCoordinator``, it parses them all.
@@ -953,9 +959,8 @@ several Java coordinators, an unmapped bundle's JARs fail to parse too.
   ``Main-Class`` are parsed. List JARs that should not be parsed in ``.airflowignore``.
 * Two JARs in the bundle that set the same ``Main-Class`` fail to parse, because the JVM would load
   the classes of only one of them. Keep one in the bundle.
-* Set ``queue`` on every task, with ``@Airflow.Task(queue = "java-native")`` or
-  ``TaskDef.config("queue", "java-native")``, so it runs on the coordinator's queue. There is no
-  Dag-level queue yet.
+* Set ``queue`` on the Dag, as :ref:`java-sdk/native-dags` shows, so its tasks run on a queue that
+  ``queue_to_coordinator`` routes to a ``JavaCoordinator``, such as ``java`` above.
 * A task runs the JAR of its Dag on the ``JavaCoordinator`` that its queue routes to,
   which need not be the one that parsed the JAR. For example, a queue can route to a coordinator that uses another JDK.
   A task whose queue routes to another kind of coordinator fails without retries. A task whose JAR is missing,

@@ -31,9 +31,9 @@
  * Bundle for the native Dag E2E tests: Dags declared entirely in Java, parsed by the Dag processor
  * from this JAR.
  *
- * <p>This is the bundle's main class, so it is the Dag source the Airflow UI shows. Every task sets
- * {@code queue}, which {@code queue_to_coordinator} routes to the {@code java-jdk} coordinator.
- * The {@code java-native} coordinator only parses this JAR.
+ * <p>This is the bundle's main class, so it is the Dag source the Airflow UI shows. Each Dag sets
+ * {@code queue} once and every task inherits it, which {@code queue_to_coordinator} routes to the
+ * {@code java-jdk} coordinator. The {@code java-native} coordinator only parses this JAR.
  */
 public class NativeBundleBuilder {
   public static final String QUEUE = "java-native";
@@ -110,34 +110,30 @@ public static DagDef buildDag() {
         new DagDef("java_native_e2e")
             .config("description", "Native Java Dag of the Airflow E2E tests")
             .config("catchup", false)
-            .config("tags", List.of("java-sdk", "native", "e2e"));
+            .config("tags", List.of("java-sdk", "native", "e2e"))
+            .config("queue", QUEUE);
 
-    var extract = dag.task("extract", Extract.class).config("queue", QUEUE);
-    var transform = dag.task("transform", Transform.class).config("queue", QUEUE);
-    var load = dag.task("load", Load.class).config("queue", QUEUE);
+    var extract = dag.task("extract", Extract.class);
+    var transform = dag.task("transform", Transform.class);
+    var load = dag.task("load", Load.class);
 
     transform.after(extract).before(load);
 
-    var reportMany = dag.task("report_many", ReportMany.class).config("queue", QUEUE);
-    var reportFew = dag.task("report_few", ReportFew.class).config("queue", QUEUE);
-    dag.If("has_rows", HasRows.class).after(transform).config("queue", QUEUE).Then(reportMany).Else(reportFew);
+    var reportMany = dag.task("report_many", ReportMany.class);
+    var reportFew = dag.task("report_few", ReportFew.class);
+    dag.If("has_rows", HasRows.class).after(transform).Then(reportMany).Else(reportFew);
 
-    var reportLong = dag.task("report_long", ReportLong.class).config("queue", QUEUE);
-    var reportShort = dag.task("report_short", ReportShort.class).config("queue", QUEUE);
-    dag.Switch("pick_report", PickReport.class)
-        .after(transform)
-        .config("queue", QUEUE)
-        .Case(reportLong)
-        .Case(reportShort);
+    var reportLong = dag.task("report_long", ReportLong.class);
+    var reportShort = dag.task("report_short", ReportShort.class);
+    dag.Switch("pick_report", PickReport.class).after(transform).Case(reportLong).Case(reportShort);
 
     // 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_e2e")).config("queue", QUEUE);
+    var trigger = dag.task("trigger_downstream", new TriggerDagRun("java_native_target_e2e"));
     load.before(trigger);
 
     // Ordering-only edge: the checks group runs after extract, with no data flowing.
     var checks = dag.taskGroup("checks");
-    checks.task("audit", Audit.class).config("queue", QUEUE);
+    checks.task("audit", Audit.class);
     extract.before(checks);
 
     return dag;

@@ -24,58 +24,59 @@
 
 import org.apache.airflow.sdk.*;
 
-/** A native Java Dag declared with annotations; its tasks use the interface Dag's queue. */
+/** A native Java Dag declared with annotations; it sets the same queue as the interface Dag. */
 @Airflow.Dag(
     id = "java_native_annotation_e2e",
     description = "Native Java Dag of the Airflow E2E tests, declared with annotations",
     catchup = false,
+    queue = QUEUE,
     tags = {"java-sdk", "native", "e2e"})
 public class AnnotationDag {
-  @Airflow.Task(id = "extract", queue = QUEUE)
+  @Airflow.Task(id = "extract")
   public long extract() {
     return 42L;
   }
 
-  @Airflow.Task(id = "transform", queue = QUEUE)
+  @Airflow.Task(id = "transform")
   public long transform(long extracted, double factor) {
     return (long) (extracted * factor);
   }
 
   /** Fails the run unless the value reached it, so a successful run proves the XCom flow. */
-  @Airflow.Task(id = "load", queue = QUEUE)
+  @Airflow.Task(id = "load")
   public void load(long transformed) {
     if (transformed != 63L) {
       throw new IllegalStateException("load expected 63 from transform, got " + transformed);
     }
   }
 
-  @Airflow.Task(id = "report_many", queue = QUEUE)
+  @Airflow.Task(id = "report_many")
   public void reportMany() {}
 
-  @Airflow.Task(id = "report_few", queue = QUEUE)
+  @Airflow.Task(id = "report_few")
   public void reportFew() {}
 
   /** Picks one of {@link #reportMany} and {@link #reportFew}; the other is skipped. */
-  @Airflow.If(id = "has_rows", queue = QUEUE)
+  @Airflow.If(id = "has_rows")
   public boolean hasRows(long transformed) {
     return transformed > 0;
   }
 
-  @Airflow.Task(id = "report_long", queue = QUEUE)
+  @Airflow.Task(id = "report_long")
   public void reportLong() {}
 
-  @Airflow.Task(id = "report_short", queue = QUEUE)
+  @Airflow.Task(id = "report_short")
   public void reportShort() {}
 
   /** Names the one of {@link #reportLong} and {@link #reportShort} that runs; the other is skipped. */
-  @Airflow.Switch(id = "pick_report", queue = QUEUE)
+  @Airflow.Switch(id = "pick_report")
   public Class<? extends Task> pickReport(long transformed) {
     return transformed > 100 ? AnnotationDagBuilder.ReportLong.class : AnnotationDagBuilder.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.
-  @Airflow.Task(id = "trigger_downstream", queue = QUEUE)
+  @Airflow.Task(id = "trigger_downstream")
   public TriggerDagRun triggerDownstream() {
     return new TriggerDagRun("java_native_target_e2e");
   }
@@ -84,7 +85,7 @@ public TriggerDagRun triggerDownstream() {
   // the task "checks.audit".
   @Airflow.TaskGroup(id = "checks")
   static class Checks {
-    @Airflow.Task(id = "audit", queue = QUEUE)
+    @Airflow.Task(id = "audit")
     public void audit() {}
   }
 

@@ -65,8 +65,9 @@ public static List<DagDef> load() throws IOException, URISyntaxException {
       var dag =
           new DagDef(dagId)
               .config("catchup", false)
-              .config("tags", List.of("java-sdk", "java_dynamic"));
-      dag.task("greet", Greet.class).config("queue", QUEUE);
+              .config("tags", List.of("java-sdk", "java_dynamic"))
+              .config("queue", QUEUE);
+      dag.task("greet", Greet.class);
       dags.add(dag);
     }
     return dags;

@@ -39,8 +39,9 @@ public static DagDef build() {
         new DagDef("java_native_target_e2e")
             .config("description", "Native Java Dag the e2e's other native Dags trigger")
             .config("catchup", false)
-            .config("tags", List.of("java-sdk", "native", "e2e"));
-    dag.task("receive", Receive.class).config("queue", QUEUE);
+            .config("tags", List.of("java-sdk", "native", "e2e"))
+            .config("queue", QUEUE);
+    dag.task("receive", Receive.class);
     return dag;
   }
 }
@@ -24,11 +24,12 @@
 
 No Python file declares these Dags. The ``airflow-e2e-tests/java-native-bundle`` JAR sits in the Dag
 bundle. There are four ``JavaCoordinator``s, so ``[sdk] dag_bundle_to_coordinator`` picks the
-``java-native`` one to parse it: the Dag processor runs the JAR through it. Each task sets
-``queue="java-native"``, which routes to ``java-jdk``. That coordinator runs the same JAR from the Dag's
-own bundle, whatever its ``task_handler_bundle_name`` says. ``java_native_e2e`` is declared with the interface
-API, ``java_native_annotation_e2e`` with annotations. Both also exercise a task group, an ``If``, a
-``Switch`` and a ``TriggerDagRun`` of ``java_native_target_e2e``, the third Dag the JAR declares.
+``java-native`` one to parse it: the Dag processor runs the JAR through it. Each Dag sets
+``queue="java-native"`` once and every task inherits it, which routes to ``java-jdk``. That
+coordinator runs the same JAR from the Dag's own bundle, whatever its ``task_handler_bundle_name``
+says. ``java_native_e2e`` is declared with the interface API, ``java_native_annotation_e2e`` with
+annotations. Both also exercise a task group, an ``If``, a ``Switch`` and a ``TriggerDagRun`` of
+``java_native_target_e2e``, the third Dag the JAR declares.
 """
 
 from __future__ import annotations
@@ -43,7 +44,7 @@
 # The Dag processor starts a JVM to parse the JAR, and each task starts another.
 _JAVA_TASK_TIMEOUT = 600
 
-# The queue the Dag's tasks set, which ``queue_to_coordinator`` routes to ``java-jdk``.
+# The queue the Dag sets for all its tasks, which ``queue_to_coordinator`` routes to ``java-jdk``.
 _QUEUE = "java-native"
 
 # The Dag ``trigger_downstream`` starts a run of; not triggered through the client itself.
@@ -149,8 +150,8 @@ def test_the_graph_is_the_one_java_declared(parsed_dags: AirflowClient, native_d
 
 
 @_by_dag_id
-def test_every_task_sets_the_native_queue(parsed_dags: AirflowClient, native_dag: _NativeDag):
-    """The Java DSL has no Dag-level queue, so each task sets its own."""
+def test_every_task_takes_the_dag_queue(parsed_dags: AirflowClient, native_dag: _NativeDag):
+    """The Dag sets ``queue`` once, and every task runs on it."""
     tasks = parsed_dags.get_tasks(native_dag.dag_id).get("tasks", [])
 
     assert {task["task_id"]: task["queue"] for task in tasks} == dict.fromkeys(native_dag.downstream, _QUEUE)

@@ -31,6 +31,7 @@
     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"})

@@ -101,6 +101,7 @@ public static DagDef build() {
         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"));
 

@@ -41,6 +41,7 @@ public static DagDef build() {
     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);

@@ -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")

@@ -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() {