jason810496 commented on code in PR #74538:
URL: https://github.com/apache/airflow/pull/74538#discussion_r4345365335
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt:
##########
@@ -156,7 +159,11 @@ private fun serializeTask(
// __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) ->
+ // The Dag's queue is merged in first, so one equal to the schema default is
left out too.
+ // A trigger task takes it as well, because the Java runtime runs it.
+ val config =
+ if (dagQueue == null || "queue" in def.configValues) def.configValues else
def.configValues + ("queue" to dagQueue)
Review Comment:
The task table carries the same flag now, so `TaskDef.config("queue", "")`
and `@Airflow.Task(queue = "")` are rejected too. An unset queue is still fine
at either level.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt:
##########
@@ -85,19 +85,24 @@ 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,
value: Any?,
): DagDef {
- dagConfig[key] = checkConfigValue("Dag", SchemaFields.DAG, key, value)
+ val checked = checkConfigValue("Dag", SchemaFields.DAG, key, value)
+ require(key != "queue" || (checked as String).isNotBlank()) { "Dag config
key 'queue' must not be blank" }
Review Comment:
Done. `Field` now carries a `notBlank` flag, generated from one
`notBlankKeys = setOf("queue")` for both the Dag and the task table, and a
shared `checkStringConstraints` enforces it. `DagDef.config` and the processor
both lost their `queue` branch and only supply the message prefix, so the two
errors stay in sync.
##########
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt:
##########
@@ -1951,6 +1952,29 @@ class AirflowProcessorTest {
)
}
+ @Test
+ @DisplayName("reject a blank Dag-level queue")
+ fun rejectBlankDagQueue() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag(queue = " ")
Review Comment:
Both blank-queue tests now loop over `""` and `" "`, plus a new task-level
one.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt:
##########
@@ -156,7 +159,11 @@ private fun serializeTask(
// __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) ->
+ // The Dag's queue is merged in first, so one equal to the schema default is
left out too.
+ // A trigger task takes it as well, because the Java runtime runs it.
+ val config =
+ if (dagQueue == null || "queue" in def.configValues) def.configValues else
def.configValues + ("queue" to dagQueue)
+ config.forEach { (key, value) ->
Review Comment:
Applied, and the comment block moved down onto the `forEach`.
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -595,6 +596,8 @@ 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. The Dag's
``queue`` cannot be blank.
Review Comment:
Wrapped both. 599 now says a `queue` that is set cannot be blank, since the
rule covers task queues too.
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt:
##########
@@ -186,6 +209,48 @@ 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",
+ // Set on the task, so it wins; at the schema default, so it is left
out.
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
