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 <noreply@anthropic.com>

* Register the interface example handlers inline

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

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"

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
 ```

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 {

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)

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);
-  }
 }

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);

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()