jason810496 commented on code in PR #74538:
URL: https://github.com/apache/airflow/pull/74538#discussion_r4347444072
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt:
##########
@@ -150,13 +153,17 @@ 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) ->
+ // The Dag's queue is merged in before this, so one equal to the schema
default is left out too.
Review Comment:
Dropped.
##########
airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py:
##########
@@ -24,8 +24,8 @@
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
+``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
Review Comment:
Rewrapped the paragraph.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Fields.kt:
##########
@@ -57,14 +57,32 @@ 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 [field] carries on a string value. [describe] names
+ * the value, so each caller keeps its own wording. Public so that the
+ * annotation processor enforces the same constraints on `@Airflow.Dag` /
+ * `@Airflow.Task` attributes; not user-facing API.
+ *
+ * @throws IllegalArgumentException if [text] breaks a constraint.
+ */
+fun checkStringConstraints(
+ field: Field,
+ text: String,
+ describe: () -> String,
+) {
+ require(!field.notBlank || text.isNotBlank()) { "${describe()} must not be
blank" }
+}
Review Comment:
Good idea, Moved onto `Field` as `checkString` method.
--
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]
