This is an automated email from the ASF dual-hosted git repository.
potiuk 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 31b558a6076 Fix Dag id collision in the Java SDK example bundle
(#76019)
31b558a6076 is described below
commit 31b558a607684d5c1939e081d5e28e39a6012869
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Tue Nov 3 10:40:54 2026 +0800
Fix Dag id collision in the Java SDK example bundle (#76019)
* Fix Dag id collision in the Java SDK example bundle
The interface example registered its task handlers as a Java Dag, so the
example jar declared a Dag with the same id as the Python Dag that owns
those tasks. When the jar is also a Dag bundle, the jar's Dag replaced
the Python one and its runs failed. The new java_task_handler_ prefix
also tells these Dags apart from the Java-defined java_native_ ones.
Co-Authored-By: Claude <[email protected]>
* Register the interface example handlers inline
Co-Authored-By: Claude <[email protected]>
---------
Co-authored-by: Claude <[email protected]>
---
.../java_sdk_tests/test_java_sdk_dag.py | 10 +++++-----
java-sdk/example/README.md | 16 ++++++++--------
.../org/apache/airflow/example/AnnotationExample.java | 10 +++++-----
.../apache/airflow/example/ExampleBundleBuilder.java | 11 ++++++++++-
.../airflow/example/InterfaceExampleBuilder.java | 7 -------
.../org/apache/airflow/example/XComCastingExample.java | 16 ++++++++--------
java-sdk/example/src/resources/dags/java_examples.py | 18 +++++++++---------
7 files changed, 45 insertions(+), 43 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 ec2f1408f91..d990b370566 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
@@ -23,7 +23,7 @@ Run with::
What is verified
----------------
-The test triggers the ``java_annotation_example`` Dag, which has this task
+The test triggers the ``java_task_handler_annotation_example`` Dag, which has
this task
graph::
python_task_1 >> extract >> transform >> [load, python_task_2]
@@ -60,7 +60,7 @@ confirms:
received, so it only succeeds if ``run_label`` reached its ``runLabel``
field.
-The ``java_interface_example`` and ``scala_spark_example`` Dags are covered
too:
+The ``java_task_handler_interface_example`` and ``scala_spark_example`` Dags
are covered too:
the former for tasks written against ``InputTask``, whose arguments the SDK
resolves from the stub call site into a ``TaskInput`` and injects.
"""
@@ -90,9 +90,9 @@ _SPARK_TASK_TIMEOUT = 1200
# Logs can lag slightly behind the task reaching a terminal state.
_LOG_FETCH_TIMEOUT = 60
-_ANNOTATION_DAG_ID = "java_annotation_example"
-_INTERFACE_DAG_ID = "java_interface_example"
-_XCOM_CASTING_DAG_ID = "java_xcom_casting_example"
+_ANNOTATION_DAG_ID = "java_task_handler_annotation_example"
+_INTERFACE_DAG_ID = "java_task_handler_interface_example"
+_XCOM_CASTING_DAG_ID = "java_task_handler_xcom_casting_example"
_SCALA_SPARK_DAG_ID = "scala_spark_example"
_VARIABLE_WRITE_DAG_ID = "java_variable_write"
_VARIABLE_WRITE_DESCRIPTION = "written by the Java SDK e2e test"
diff --git a/java-sdk/example/README.md b/java-sdk/example/README.md
index 4812787146b..2bf58ec6d21 100644
--- a/java-sdk/example/README.md
+++ b/java-sdk/example/README.md
@@ -79,11 +79,11 @@ airflow dags trigger java_native_interface_example
These Dags are declared in Python, in `src/resources/dags/java_examples.py`,
and their stub tasks are
implemented in Java:
-| Dag | Source | Shows
|
-| --------------------------- | ------------------------------ |
---------------------------------------------------------------------- |
-| `java_annotation_example` | `AnnotationExample.java` |
`@Airflow.TaskHandler` methods, with arguments by position and by name |
-| `java_interface_example` | `InterfaceExampleBuilder.java` | `Task` and
`InputTask` classes registered by Dag and task id |
-| `java_xcom_casting_example` | `XComCastingExample.java` | How values
passed between tasks convert to the declared Java types |
+| Dag | Source |
Shows |
+| ---------------------------------------- | ------------------------------ |
---------------------------------------------------------------------- |
+| `java_task_handler_annotation_example` | `AnnotationExample.java` |
`@Airflow.TaskHandler` methods, with arguments by position and by name |
+| `java_task_handler_interface_example` | `InterfaceExampleBuilder.java` |
`Task` and `InputTask` classes registered by Dag and task id |
+| `java_task_handler_xcom_casting_example` | `XComCastingExample.java` |
How values passed between tasks convert to the declared Java types |
Copy `src/resources/dags/java_examples.py` into your Airflow Dags folder, and
create the Connection and
Variable the examples read:
@@ -101,7 +101,7 @@ export AIRFLOW_VAR_MY_VARIABLE=123
```
```bash
-airflow dags trigger java_annotation_example
-airflow dags trigger java_interface_example
-airflow dags trigger java_xcom_casting_example
+airflow dags trigger java_task_handler_annotation_example
+airflow dags trigger java_task_handler_interface_example
+airflow dags trigger java_task_handler_xcom_casting_example
```
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
b/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
index 2a9059f91ea..0b36b222743 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/AnnotationExample.java
@@ -34,7 +34,7 @@ import org.apache.airflow.sdk.*;
public class AnnotationExample {
private static final System.Logger log =
System.getLogger(AnnotationExample.class.getName());
- @Airflow.TaskHandler(dag = "java_annotation_example", task = "extract")
+ @Airflow.TaskHandler(dag = "java_task_handler_annotation_example", task =
"extract")
public long extractValue(Client client) throws InterruptedException {
log.log(INFO, "Hello from task");
@@ -53,7 +53,7 @@ public class AnnotationExample {
return new Date().getTime();
}
- @Airflow.TaskHandler(dag = "java_annotation_example", task = "transform")
+ @Airflow.TaskHandler(dag = "java_task_handler_annotation_example", task =
"transform")
public long transformValue(Client client, long extracted) {
log.log(INFO, "Got XCom from extract: {0}", extracted);
@@ -69,7 +69,7 @@ public class AnnotationExample {
// the task UP_FOR_RETRY -- which only works because the Java SDK now returns
// RetryTask (instead of a terminal FAILED) when ti_context.should_retry is
// set. The retry then runs this task again and it returns normally.
- @Airflow.TaskHandler(dag = "java_annotation_example")
+ @Airflow.TaskHandler(dag = "java_task_handler_annotation_example")
public void load(Context context, long transformed) {
log.log(INFO, "Got XCom from transform: {0}", transformed);
if (context.ti.tryNumber == 1) {
@@ -86,7 +86,7 @@ public class AnnotationExample {
public long transformed;
}
- @Airflow.TaskHandler(dag = "java_annotation_example", task = "report")
+ @Airflow.TaskHandler(dag = "java_task_handler_annotation_example", task =
"report")
public void report(ReportInput input) {
log.log(INFO, "Report {0} for transformed value {1}", input.runLabel,
input.transformed);
if (!"nightly".equals(input.runLabel)) {
@@ -95,7 +95,7 @@ public class AnnotationExample {
}
// Verify one supervisor channel can handle client calls across threads.
- @Airflow.TaskHandler(dag = "java_annotation_example", task = "concurrent")
+ @Airflow.TaskHandler(dag = "java_task_handler_annotation_example", task =
"concurrent")
public void concurrentClientCalls(Client client) throws Exception {
var pool = Executors.newFixedThreadPool(8);
try {
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 30da2aa8e11..7700ab0da03 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
@@ -26,7 +26,16 @@ import org.apache.airflow.sdk.*;
public class ExampleBundleBuilder {
public static Bundle build() {
return new Bundle()
- .register(InterfaceExampleBuilder.build())
+ .register(
+ "java_task_handler_interface_example", "extract",
InterfaceExampleBuilder.Extract.class)
+ .register(
+ "java_task_handler_interface_example",
+ "transform",
+ InterfaceExampleBuilder.Transform.class)
+ .register(
+ "java_task_handler_interface_example",
+ "summarize",
+ InterfaceExampleBuilder.Summarize.class)
.register(AnnotationExample.class)
.register(XComCastingExample.class)
.register(org.apache.airflow.example.nativedag.AnnotationExample.class)
diff --git
a/java-sdk/example/src/java/org/apache/airflow/example/InterfaceExampleBuilder.java
b/java-sdk/example/src/java/org/apache/airflow/example/InterfaceExampleBuilder.java
index 4d0a494d6dc..80247b61be2 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/InterfaceExampleBuilder.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/InterfaceExampleBuilder.java
@@ -88,11 +88,4 @@ public class InterfaceExampleBuilder {
}
}
}
-
- public static DagDef build() {
- return new DagDef("java_interface_example")
- .addTask("extract", Extract.class)
- .addTask("transform", Transform.class)
- .addTask("summarize", Summarize.class);
- }
}
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
index f3608fbd1b3..339bbaeb4b5 100644
---
a/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
+++
b/java-sdk/example/src/java/org/apache/airflow/example/XComCastingExample.java
@@ -30,20 +30,20 @@ import org.apache.airflow.sdk.*;
public class XComCastingExample {
private static final System.Logger log =
System.getLogger(XComCastingExample.class.getName());
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_number")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"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.
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_long")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"widen_to_long")
public long widenToLong(long value) {
log.log(INFO, "Got long {0}", value);
return value + 1;
}
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"widen_to_double")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"widen_to_double")
public void widenToDouble(double value) {
log.log(INFO, "Got double {0}", value);
if (value != 8.0) {
@@ -51,12 +51,12 @@ public class XComCastingExample {
}
}
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_nothing")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"produce_nothing")
public void produceNothing() {
// Pushes no return_value XCom.
}
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_nullable")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"consume_nullable")
public void consumeNullable(Integer value) {
log.log(INFO, "Got nullable int {0}", value);
if (value != null) {
@@ -64,13 +64,13 @@ public class XComCastingExample {
}
}
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"produce_fraction")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"produce_fraction")
public double produceFraction() {
log.log(INFO, "Producing double 1.5");
return 1.5;
}
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_float")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"consume_float")
public void consumeFloat(float value) {
log.log(INFO, "Got float {0}", value);
if (value != 1.5f) {
@@ -81,7 +81,7 @@ public class XComCastingExample {
// A parameter with type arguments keeps its element type. Reading an element
// is what would fail if it did not: the wire integers decode to Long unless
// the declared List<Double> survives the binding.
- @Airflow.TaskHandler(dag = "java_xcom_casting_example", task =
"consume_double_list")
+ @Airflow.TaskHandler(dag = "java_task_handler_xcom_casting_example", task =
"consume_double_list")
public void consumeDoubleList(List<Double> values) {
log.log(INFO, "Got list {0}", values);
double total = values.get(0) + values.get(1);
diff --git a/java-sdk/example/src/resources/dags/java_examples.py
b/java-sdk/example/src/resources/dags/java_examples.py
index c948d2a5652..53131d8733f 100644
--- a/java-sdk/example/src/resources/dags/java_examples.py
+++ b/java-sdk/example/src/resources/dags/java_examples.py
@@ -93,8 +93,8 @@ def python_task_2(transformed):
print(transformed)
-@dag(dag_id="java_interface_example")
-def java_interface_example():
+@dag(dag_id="java_task_handler_interface_example")
+def java_task_handler_interface_example():
extracted = extract()
python_task_1() >> extracted
transformed = transform(extracted)
@@ -102,8 +102,8 @@ def java_interface_example():
summarize(region_code="emea", transformed=transformed)
-@dag(dag_id="java_annotation_example")
-def java_annotation_example():
+@dag(dag_id="java_task_handler_annotation_example")
+def java_task_handler_annotation_example():
extracted = extract()
python_task_1() >> extracted
transformed = transform(extracted)
@@ -113,14 +113,14 @@ def java_annotation_example():
concurrent()
-@dag(dag_id="java_xcom_casting_example")
-def java_xcom_casting_example():
+@dag(dag_id="java_task_handler_xcom_casting_example")
+def java_task_handler_xcom_casting_example():
widen_to_double(widen_to_long(produce_number()))
consume_nullable(produce_nothing())
consume_float(produce_fraction())
consume_double_list([1, 2])
-java_interface_example()
-java_annotation_example()
-java_xcom_casting_example()
+java_task_handler_interface_example()
+java_task_handler_annotation_example()
+java_task_handler_xcom_casting_example()
