This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new fa9e5c7aaca Fix Java SDK numeric XCom casting and remove unused XCom
key field (#68983)
fa9e5c7aaca is described below
commit fa9e5c7aacac4966abc5f8dd24447c6fc2cc9530
Author: PoAn Yang <[email protected]>
AuthorDate: Thu Jul 2 14:34:19 2026 +0900
Fix Java SDK numeric XCom casting and remove unused XCom key field (#68983)
* Fix Java SDK numeric XCom casting
Generate accessors that widen through Number instead: primitive params use
((Number) x).intValue(), and boxed params use
Optional.ofNullable(...).orElse(null)
so an absent XCom stays null. Add a java_xcom_casting_example DAG and
processor
tests covering int->long->double widening and an absent boxed value.
Signed-off-by: PoAn Yang <[email protected]>
* Remove unused key field from Java SDK Builder.XCom annotation
Signed-off-by: PoAn Yang <[email protected]>
* Add float cross-type coverage to Java SDK XCom casting tests
Signed-off-by: PoAn Yang <[email protected]>
* Remove unused comment
Signed-off-by: PoAn Yang <[email protected]>
---------
Signed-off-by: PoAn Yang <[email protected]>
---
.../java_sdk_tests/test_java_sdk_dag.py | 26 ++++++++
.../airflow/example/ExampleBundleBuilder.java | 5 +-
.../apache/airflow/example/XComCastingExample.java | 77 ++++++++++++++++++++++
.../example/src/resources/dags/java_examples.py | 36 ++++++++++
.../org/apache/airflow/sdk/BuilderProcessor.kt | 46 +++++++++++--
.../kotlin/org/apache/airflow/sdk/BuilderTest.kt | 65 +++++++++++++++++-
.../main/kotlin/org/apache/airflow/sdk/Builder.kt | 2 -
7 files changed, 246 insertions(+), 11 deletions(-)
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dag.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dag.py
index 867fb098aa3..e634ece89b7 100644
---
a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dag.py
+++
b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dag.py
@@ -165,6 +165,32 @@ class TestJavaSDKAnnotationExample:
f" all tasks : { {k: v.get('state') for k, v in ti_map.items()}
}"
)
+ def test_numeric_xcom_casting(self):
+ """Numeric XComs are read across declared types (int -> long ->
double, and a wire
+ double back as a float), and a boxed param stays null when its XCom is
absent."""
+ resp = self.airflow_client.trigger_dag(
+ "java_xcom_casting_example",
+ json={"logical_date": datetime.now(timezone.utc).isoformat()},
+ )
+ run_id = resp["dag_run_id"]
+
+ dag_state = self.airflow_client.wait_for_dag_run(
+ dag_id="java_xcom_casting_example",
+ run_id=run_id,
+ timeout=_JAVA_TASK_TIMEOUT,
+ )
+
+ ti_resp =
self.airflow_client.get_task_instances(dag_id="java_xcom_casting_example",
run_id=run_id)
+ ti_map = {ti["task_id"]: ti for ti in ti_resp.get("task_instances",
[])}
+
+ for task_id in ("widen_to_double", "consume_nullable",
"consume_float"):
+ assert ti_map.get(task_id, {}).get("state") == "success", (
+ f"Java {task_id!r} task did not succeed.\n"
+ f" task state : {ti_map.get(task_id, {}).get('state')!r}\n"
+ f" dag state : {dag_state!r}\n"
+ f" all tasks : { {k: v.get('state') for k, v in
ti_map.items()} }"
+ )
+
def test_load_retried_then_succeeded(self):
"""``load`` fails once (UP_FOR_RETRY) then succeeds on the second
attempt.
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java
b/java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java
index 7c8d9955dc0..f63d6c4d743 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java
@@ -27,7 +27,10 @@ public class ExampleBundleBuilder implements BundleBuilder {
@NotNull
@Override
public Iterable<Dag> getDags() {
- return List.of(InterfaceExampleBuilder.build(),
AnnotationExampleBuilder.build());
+ return List.of(
+ InterfaceExampleBuilder.build(),
+ AnnotationExampleBuilder.build(),
+ XComCastingExampleBuilder.build());
}
public static void main(String[] args) {
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
b/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
new file mode 100644
index 00000000000..c4945af865d
--- /dev/null
+++
b/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
@@ -0,0 +1,77 @@
+/*
+ * 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.example;
+
+import static java.lang.System.Logger.Level.INFO;
+
+import org.apache.airflow.sdk.*;
+
[email protected](id = "java_xcom_casting_example")
+public class XComCastingExample {
+ private static final System.Logger log =
System.getLogger(XComCastingExample.class.getName());
+
+ @Builder.Task(id = "produce_number")
+ public int produceNumber() {
+ log.log(INFO, "Producing int 7");
+ return 7;
+ }
+
+ // Any primitive numeric type (byte, short, int, long, float, double) and
its boxed form works the same way.
+ @Builder.Task(id = "widen_to_long")
+ public long widenToLong(@Builder.XCom(task = "produce_number") long value) {
+ log.log(INFO, "Got long {0}", value);
+ return value + 1;
+ }
+
+ @Builder.Task(id = "widen_to_double")
+ public void widenToDouble(@Builder.XCom(task = "widen_to_long") double
value) {
+ log.log(INFO, "Got double {0}", value);
+ if (value != 8.0) {
+ throw new RuntimeException("expected 8.0 but got " + value);
+ }
+ }
+
+ @Builder.Task(id = "produce_nothing")
+ public void produceNothing() {
+ // Pushes no return_value XCom.
+ }
+
+ @Builder.Task(id = "consume_nullable")
+ public void consumeNullable(@Builder.XCom(task = "produce_nothing") Integer
value) {
+ log.log(INFO, "Got nullable int {0}", value);
+ if (value != null) {
+ throw new RuntimeException("expected null but got " + value);
+ }
+ }
+
+ @Builder.Task(id = "produce_fraction")
+ public double produceFraction() {
+ log.log(INFO, "Producing double 1.5");
+ return 1.5;
+ }
+
+ @Builder.Task(id = "consume_float")
+ public void consumeFloat(@Builder.XCom(task = "produce_fraction") float
value) {
+ log.log(INFO, "Got float {0}", value);
+ if (value != 1.5f) {
+ throw new RuntimeException("expected 1.5 but got " + value);
+ }
+ }
+}
diff --git a/java-sdk/example/src/resources/dags/java_examples.py
b/java-sdk/example/src/resources/dags/java_examples.py
index 38be0e9e947..5426911b0fa 100644
--- a/java-sdk/example/src/resources/dags/java_examples.py
+++ b/java-sdk/example/src/resources/dags/java_examples.py
@@ -45,6 +45,34 @@ def load(): ...
def concurrent(): ...
[email protected](queue="java")
+def produce_number(): ...
+
+
[email protected](queue="java")
+def widen_to_long(): ...
+
+
[email protected](queue="java")
+def widen_to_double(): ...
+
+
[email protected](queue="java")
+def produce_nothing(): ...
+
+
[email protected](queue="java")
+def consume_nullable(): ...
+
+
[email protected](queue="java")
+def produce_fraction(): ...
+
+
[email protected](queue="java")
+def consume_float(): ...
+
+
@task()
def python_task_2(transformed):
print("python_task_2")
@@ -68,5 +96,13 @@ def java_annotation_example():
concurrent()
+@dag(dag_id="java_xcom_casting_example")
+def java_xcom_casting_example():
+ produce_number() >> widen_to_long() >> widen_to_double()
+ produce_nothing() >> consume_nullable()
+ produce_fraction() >> consume_float()
+
+
java_interface_example()
java_annotation_example()
+java_xcom_casting_example()
diff --git
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
index 979615c12f2..b764c00a994 100644
---
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
+++
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
@@ -22,10 +22,12 @@
package org.apache.airflow.sdk
import com.squareup.javapoet.ClassName
+import com.squareup.javapoet.CodeBlock
import com.squareup.javapoet.JavaFile
import com.squareup.javapoet.MethodSpec
import com.squareup.javapoet.TypeName
import com.squareup.javapoet.TypeSpec
+import java.util.Optional
import javax.annotation.processing.AbstractProcessor
import javax.annotation.processing.ProcessingEnvironment
import javax.annotation.processing.RoundEnvironment
@@ -161,12 +163,7 @@ class BuilderProcessor : AbstractProcessor() {
}
}
required.forEach {
- executeSpec.addStatement(
- $$"var $L = ($T) client.getXCom($S)",
- it.paramName,
- with(TypeName.get(it.paramType)) { if (isPrimitive) box() else this },
- it.taskId,
- )
+ executeSpec.addStatement($$"var $L = $L", it.paramName, xcomAccess(it))
}
if (inner.returnType.kind == TypeKind.VOID) {
$$"new $T().$L($L)"
@@ -203,6 +200,43 @@ private data class RequiredXCom(
val taskId: String,
)
+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 = CodeBlock.of($$"client.getXCom($S)", xcom.taskId)
+ val type = TypeName.get(xcom.paramType)
+ val accessor = NUMBER_ACCESSORS[type]
+ val number = ClassName.get(Number::class.java)
+ // Wire integers decode to Long and floats to Double, so a direct
(Integer)/(Float)
+ // cast throws ClassCastException; widen via Number instead.
+ return when {
+ accessor == null -> CodeBlock.of($$"($T) $L", if (type.isPrimitive)
type.box() else type, call)
+ type.isPrimitive -> CodeBlock.of($$"(($T) $L).$L()", number, call,
accessor)
+ else ->
+ CodeBlock.of(
+ $$"$T.ofNullable(($T) $L).map($T::$L).orElse(null)",
+ ClassName.get(Optional::class.java),
+ number,
+ call,
+ number,
+ accessor,
+ )
+ }
+}
+
private data class BuildTaskResult(
val spec: TypeSpec,
)
diff --git
a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
index bd3aae81d13..692374d9e3c 100644
--- a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
+++ b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
@@ -69,6 +69,7 @@ class BuilderTest {
""",
)
+ assertThat(compilation).succeeded()
assertThat(compilation)
.generatedSourceFile("org.apache.airflow.example.TestExampleBuilder")
.hasSourceEquivalentTo(
@@ -77,7 +78,7 @@ class BuilderTest {
package org.apache.airflow.example;
import java.lang.Exception;
- import java.lang.Integer;
+ import java.lang.Number;
import java.lang.Override;
import org.apache.airflow.sdk.Client;
import org.apache.airflow.sdk.Context;
@@ -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);
}
}
@@ -116,6 +117,66 @@ class BuilderTest {
)
}
+ @Test
+ @DisplayName("widen primitive numerics directly and boxed numerics
null-safely")
+ fun generateBuilderWidensNumericXCom() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Builder;
+ @Builder.Dag
+ public class TestExample {
+ @Builder.Task
+ public void t(
+ @Builder.XCom(task = "a") int i,
+ @Builder.XCom(task = "b") long l,
+ @Builder.XCom(task = "c") double d,
+ @Builder.XCom(task = "f") float fl,
+ @Builder.XCom(task = "e") Integer boxed) {}
+ }
+ """,
+ )
+
+ assertThat(compilation).succeeded()
+ assertThat(compilation)
+ .generatedSourceFile("org.apache.airflow.example.TestExampleBuilder")
+ .hasSourceEquivalentTo(
+ "org.apache.airflow.example.TestExampleBuilder",
+ """
+ package org.apache.airflow.example;
+
+ import java.lang.Exception;
+ import java.lang.Number;
+ import java.lang.Override;
+ import java.util.Optional;
+ import org.apache.airflow.sdk.Client;
+ import org.apache.airflow.sdk.Context;
+ import org.apache.airflow.sdk.Dag;
+ import org.apache.airflow.sdk.Task;
+
+ public final class TestExampleBuilder {
+ public static Dag build() {
+ var dag = new Dag("TestExample");
+ dag.addTask("t", T.class);
+ return dag;
+ }
+ public static final class T implements Task {
+ @Override
+ public void execute(Context context, Client client) throws
Exception {
+ var i = ((Number) client.getXCom("a")).intValue();
+ var l = ((Number) client.getXCom("b")).longValue();
+ var d = ((Number) client.getXCom("c")).doubleValue();
+ var fl = ((Number) client.getXCom("f")).floatValue();
+ var boxed = Optional.ofNullable((Number)
client.getXCom("e")).map(Number::intValue).orElse(null);
+ new TestExample().t(i, l, d, fl, boxed);
+ }
+ }
+ }
+ """,
+ )
+ }
+
@Test
@DisplayName("generate builder for dag class with custom dag id")
fun generateBuilderWithCustomDagId() {
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Builder.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Builder.kt
index c12a53f83e5..3a5b84d2daf 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Builder.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Builder.kt
@@ -79,12 +79,10 @@ class Builder internal constructor() {
*
* @param task The task ID to pull. If empty or not given, the annotated
* parameter's name is used by default.
- * @param key The XCom key to pull. Defaults to the task's return value.
*/
@Target(AnnotationTarget.VALUE_PARAMETER)
@MustBeDocumented
annotation class XCom(
val task: String = "",
- val key: String = Client.XCOM_RETURN_KEY,
)
}
