jason810496 commented on code in PR #68983:
URL: https://github.com/apache/airflow/pull/68983#discussion_r3474744904
##########
java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java:
##########
@@ -72,4 +72,18 @@ public void load(Context context, @Builder.XCom(task =
"transform") long transfo
}
log.log(INFO, "Recovered on retry, try number {0}", context.ti.tryNumber);
}
+
+ @Builder.Task(id = "produce_int")
+ public int produceInt() {
+ log.log(INFO, "Producing an int XCom");
+ return 7;
+ }
+
+ @Builder.Task(id = "consume_int")
+ public void consumeInt(@Builder.XCom(task = "produce_int") int value) {
+ log.log(INFO, "Got int XCom: {0}", value);
+ if (value != 7) {
+ throw new RuntimeException("expected 7 but got " + value);
+ }
+ }
Review Comment:
How about introducing another new Dag definition (e.g. xcom numeric casting)
to showcase and exercise that we support `Task A: int -> Task B: Long -> Task
C: double` should success instead of adding too much task within the existing
example Dag.
Addtionally, the new Dag we're going to introduce can also exercise the
custom xcom key feature that we fix in this PR.
##########
java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt:
##########
@@ -107,7 +108,7 @@ class BuilderTest {
public static final class T3 implements Task {
@Override
public void execute(Context context, Client client) throws
Exception {
- var value = (Integer) client.getXCom("t2");
+ var value = ((Number) client.getXCom("t2")).intValue();
new TestExample().t3(context, value);
}
Review Comment:
It would be nice to add the test cases that cover the cross type casting,
thanks.
##########
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt:
##########
@@ -201,8 +197,42 @@ private data class RequiredXCom(
val paramType: TypeMirror,
val paramName: String,
val taskId: String,
+ val key: String,
)
+// Default xcom key like Client.XCOM_RETURN_KEY
+private const val XCOM_RETURN_KEY = "return_value"
+
+private val NUMBER_ACCESSORS: Map<TypeName, String> =
+ buildMap {
+ mapOf(
+ TypeName.BYTE to "byteValue",
+ TypeName.SHORT to "shortValue",
+ TypeName.INT to "intValue",
+ TypeName.LONG to "longValue",
+ TypeName.FLOAT to "floatValue",
+ TypeName.DOUBLE to "doubleValue",
+ ).forEach { (primitive, accessor) ->
+ put(primitive, accessor)
+ put(primitive.box(), accessor)
+ }
+ }
+
+private fun xcomAccess(xcom: RequiredXCom): CodeBlock {
+ val call =
+ if (xcom.key == XCOM_RETURN_KEY) {
+ CodeBlock.of($$"client.getXCom($S)", xcom.taskId)
+ } else {
+ CodeBlock.of($$"client.getXCom($S, $S)", xcom.key, xcom.taskId)
+ }
+ val type = TypeName.get(xcom.paramType)
+ // Wire integers decode to Long and floats to Double, so a direct (Integer)
or
+ // (Float) cast on the boxed value throws ClassCastException; widen via
Number.
+ return NUMBER_ACCESSORS[type]?.let { accessor ->
Review Comment:
Dropping the explicit `null` semantic / null-safety issue was caught by
Claude and it makes sense to me. Duplicate the full context here for further
reference.
---
**Boxed numeric params now NullPointerException on a null XCom (was `null`)**
`NUMBER_ACCESSORS` keys on both the primitive and boxed `TypeName`s, so a
param declared `Integer` (rather than `int`) also goes through `((Number)
client.getXCom(...)).intValue()`. When the upstream XCom is absent, `getXCom`
returns `null` and the accessor NullPointerExceptions, whereas before this PR
`(Integer) null` evaluated to `null`. For a *primitive* `int` param the
behavior is unchanged (both old and new NullPointerException on unboxing), so
this only affects boxed numeric params.
That may be acceptable "fail fast," but it's a silent behavior change worth
a deliberate call rather than a side effect of the cast fix. If we want to keep
the old `null` semantics, we can split the numeric branch by `isPrimitive` and
make only the boxed path null-safe:
```kotlin
return when {
accessor == null ->
CodeBlock.of($$"($T) $L", if (type.isPrimitive) type.box() else type,
call)
// Primitive target can't hold null; an absent XCom fails fast on
unboxing, as before.
type.isPrimitive ->
CodeBlock.of($$"(($T) $L).$L()", number, call, accessor)
// Boxed numeric target must keep null instead of NullPointerException-ing
inside the accessor.
else ->
CodeBlock.of(
$$"$T.ofNullable(($T) $L).map($T::$L).orElse(null)",
ClassName.get(Optional::class.java), number, call, number, accessor,
)
}
```
which generates, for a boxed `Integer`:
```java
var value = Optional.ofNullable((Number)
client.getXCom("t2")).map(Number::intValue).orElse(null);
```
The primitive `int` path stays `((Number) client.getXCom("t2")).intValue()`,
so the existing test is unaffected. If you take this route, a small processor
test over a boxed `Integer` param asserting the null-safe form would lock it in.
--
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]
