jason810496 commented on code in PR #75542:
URL: https://github.com/apache/airflow/pull/75542#discussion_r4435755371
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -719,6 +719,53 @@ The ``Client``
A value from a secrets backend, such as an ``AIRFLOW_VAR_*`` environment
variable, still wins over the stored
value when the Variable is read back. ``setVariable`` without a description
clears the existing description.
+The Dag run conf
+~~~~~~~~~~~~~~~~
+
+``context.dagRun.conf`` is the conf the Dag run was triggered with. It is a
``Map``, and its values have the Java
+types listed in :ref:`the XCom type mapping <java-sdk/types>`.
``context.dagRun.getConf(type)`` reads the conf into
+an instance of a class with typed fields instead:
+
+.. code-block:: java
+
+ public static class EtlConf {
+ public String targetTable;
+ public int limit = 100;
+ public LocalDate since;
+ }
+
+ public class Load implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ EtlConf etl = context.dagRun.getConf(EtlConf.class);
+ }
+ }
+
+For a run triggered with the conf ``{"target_table": "sales", "since":
"2024-01-31"}``, ``etl.targetTable`` is
+``"sales"``, ``etl.since`` is ``LocalDate.of(2024, 1, 31)``, and ``etl.limit``
stays ``100``.
+
+* Each key goes to the field, setter or record component with the same name
when case and underscores are
+ ignored, so ``target_table`` reaches ``targetTable``. Objects nested in the
conf follow the same rule. The keys
+ of a ``Map`` field stay as written.
+* ``@ArgName`` has no effect on ``getConf``. Read a key that no Java name can
match, such as ``target-table``,
+ from ``context.dagRun.conf``.
+* Keys that match no field are ignored, because other tasks in the run may
read them. A field with no matching
+ key keeps its Java default.
+* The conf holds only the values that the run was triggered with. In a Python
task, ``params`` combines the conf
+ with the defaults of the Dag's params. ``getConf`` does not add those
defaults, so in a scheduled run, every
+ field keeps its Java default.
+* The class needs a no-argument constructor and public fields or setters, so a
nested class must be ``static``.
+ A record also works. In Kotlin, declare the fields with ``@JvmField``, so
that a property such as
+ ``isEnabled`` keeps its name.
+* ``getConf`` throws ``IllegalArgumentException`` when a value cannot be
converted to the type of its field. It
+ also throws when two keys, such as ``target_table`` and ``TARGET_TABLE``,
match the same field after case and
+ underscores are ignored but neither key matches the field name exactly. A
primitive field cannot hold ``null``,
+ so ``getConf`` throws the same exception when a key holds ``null`` for a
primitive field. To let a field
+ receive ``null``, declare it with a boxed type, such as ``Integer``.
+* ``getConf`` also throws ``IllegalArgumentException`` when the class has two
fields whose names differ only in
+ case or underscores, such as ``targetTable`` and ``target_table``. It checks
the class even when the conf is
+ empty, so a task that passes such a class fails in a scheduled run too.
Review Comment:
It seems we could unify both statement.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ConfBinding.kt:
##########
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.internal
+
+import com.fasterxml.jackson.databind.DeserializationFeature
+import com.fasterxml.jackson.databind.JavaType
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.introspect.BeanPropertyDefinition
+import com.fasterxml.jackson.databind.json.JsonMapper
+import java.util.concurrent.ConcurrentHashMap
+
+// Other tasks in the Dag run may read keys that the target class does not
+// declare, so this mapper must not fail on an unknown key. By default, Jackson
+// moves a date-time that carries an offset to UTC. For example,
+// 2024-01-31T02:00+08:00 would become 2024-01-30T18:00Z and fall on another
+// date. This mapper keeps the offset as written instead.
+private val confMapper: ObjectMapper by lazy {
+ JsonMapper
+ .builder()
+ .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+ .disable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE)
Review Comment:
Nested primitive nulls silently become 0: `{"counts": [1, null]}` into an
`int[]` field gives `[1, 0]`.
```suggestion
.disable(DeserializationFeature.ADJUST_DATES_TO_CONTEXT_TIME_ZONE)
.enable(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES)
```
The check in `matchFields` only fires for a top-level primitive field, so it
misses elements of an `int[]` or of a primitive-typed collection.
`convertValue` wraps the resulting `MismatchedInputException` as
`IllegalArgumentException`, so the documented exception type still holds. Worth
a test, and the comment above should mention the third feature.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ConfBinding.kt:
##########
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.internal
+
+import com.fasterxml.jackson.databind.DeserializationFeature
+import com.fasterxml.jackson.databind.JavaType
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.databind.introspect.BeanPropertyDefinition
+import com.fasterxml.jackson.databind.json.JsonMapper
+import java.util.concurrent.ConcurrentHashMap
+
+// Other tasks in the Dag run may read keys that the target class does not
+// declare, so this mapper must not fail on an unknown key. By default, Jackson
+// moves a date-time that carries an offset to UTC. For example,
+// 2024-01-31T02:00+08:00 would become 2024-01-30T18:00Z and fall on another
+// date. This mapper keeps the offset as written instead.
+private val confMapper: ObjectMapper by lazy {
Review Comment:
Follow-up: `ArgValues.decode` keeps `ADJUST_DATES_TO_CONTEXT_TIME_ZONE` on,
so task args still shift dates. Worth one shared binder.
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -719,6 +719,53 @@ The ``Client``
A value from a secrets backend, such as an ``AIRFLOW_VAR_*`` environment
variable, still wins over the stored
value when the Variable is read back. ``setVariable`` without a description
clears the existing description.
+The Dag run conf
+~~~~~~~~~~~~~~~~
+
+``context.dagRun.conf`` is the conf the Dag run was triggered with. It is a
``Map``, and its values have the Java
+types listed in :ref:`the XCom type mapping <java-sdk/types>`.
``context.dagRun.getConf(type)`` reads the conf into
+an instance of a class with typed fields instead:
+
+.. code-block:: java
+
+ public static class EtlConf {
+ public String targetTable;
+ public int limit = 100;
+ public LocalDate since;
+ }
+
+ public class Load implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ EtlConf etl = context.dagRun.getConf(EtlConf.class);
+ }
+ }
+
+For a run triggered with the conf ``{"target_table": "sales", "since":
"2024-01-31"}``, ``etl.targetTable`` is
+``"sales"``, ``etl.since`` is ``LocalDate.of(2024, 1, 31)``, and ``etl.limit``
stays ``100``.
+
+* Each key goes to the field, setter or record component with the same name
when case and underscores are
+ ignored, so ``target_table`` reaches ``targetTable``. Objects nested in the
conf follow the same rule. The keys
+ of a ``Map`` field stay as written.
+* ``@ArgName`` has no effect on ``getConf``. Read a key that no Java name can
match, such as ``target-table``,
+ from ``context.dagRun.conf``.
Review Comment:
Does `@JsonProperty("target-table")` already work here?
`BeanPropertyDefinition.name` is the renamed one, so I think it does.
```suggestion
* ``@ArgName`` has no effect on ``getConf``. Pin a key that no Java name can
match, such as
``target-table``, with ``@JsonProperty``, or read it from
``context.dagRun.conf``.
```
Traced through `findFields` but not run, so please confirm with a test
before taking the suggestion.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt:
##########
@@ -92,7 +94,45 @@ data class DagRun(
@JvmField val runAfter: OffsetDateTime?,
@JvmField val runType: DagRunType?,
@JvmField val conf: Map<String, Any?>,
-)
+) {
+ /**
+ * Reads [conf] into a new instance of [type]. For example, [type] can be a
+ * class with one field per conf key.
+ *
+ * ```java
+ * public static class EtlConf {
+ * public String targetTable; // set from the "target_table" key
+ * public int limit = 100; // stays 100 when the conf has no "limit" key
+ * }
+ *
+ * EtlConf etl = context.dagRun.getConf(EtlConf.class);
+ * ```
+ *
+ * Each key goes to the field, setter or record component whose name matches
+ * the key when case and underscores are ignored. If one key matches a name
+ * exactly, that key wins over keys that match the same name only when case
+ * and underscores are ignored. Objects nested in the conf follow the same
+ * rules, including objects in lists and objects that are map values. The
keys
+ * of a map field stay as written.
+ *
+ * Keys that match no field are ignored, because other tasks in the run may
+ * read them. A field with no matching key keeps its default value. The class
+ * needs a no-argument constructor and public fields or setters. A record
also
+ * works. In Kotlin, declare the fields with `@JvmField`, so that a property
+ * such as `isEnabled` keeps its name.
+ *
+ * Annotating a field with [ArgName] has no effect here. Read a key that no
+ * Java name can match, such as `target-table`, from [conf] instead.
+ *
+ * @param type Class to read the conf into.
+ * @throws IllegalArgumentException if a value cannot be converted to the
type
+ * of its field, or if a key holds null for a primitive field. It is also
+ * thrown if two keys match the same field only when case and underscores
+ * are ignored, or if two fields of a class differ only in case or
+ * underscores.
+ */
+ fun <T : Any> getConf(type: Class<T>): T = bindConf(conf, type)
Review Comment:
Follow-up: no generic-type overload, no raw-JSON accessor, and `getXCom`
still has no typed read.
--
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]
