uranusjr commented on code in PR #74538:
URL: https://github.com/apache/airflow/pull/74538#discussion_r4344073671
##########
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 Dag queue rejects `""` and `" "`, but `TaskDef.config("queue", "")` and
`@Airflow.Task(queue = "")` still go through and serialize the queue value.
This means a task overriding the Dag queue with a blank value routes nowhere.
If blank is invalid, it’s invalid for both. (This can be also fixed by unifying
the check in DagDef and AirflowProcessor.)
--
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]
