Java SDK: Clearer error when a task class cannot be instantiated

airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst

@@ -205,6 +205,16 @@ Interface-based API
 Implement the ``Task`` interface directly for full control over how tasks are registered and how XComs are
 read.
 
+The runner creates a fresh instance of the task class through reflection for every task-instance run,
+which puts three constraints on the class. It must be:
+
+* Concrete: not abstract and not an interface.
+* A public no-argument constructor.
+* If nested inside another class, it must be ``static`` class.
+
+A class that violates any of these fails at runtime with a ``Cannot instantiate task class`` error in the
+task log.
+
 .. code-block:: java
 
     import org.apache.airflow.sdk.*;
@@ -218,11 +228,21 @@ read.
       }
     }
 
-Register tasks manually in a ``BundleBuilder``:
+Register tasks manually in a ``BundleBuilder``. A task class can be top-level like ``FetchTask``, or
+nested ``static`` class like ``ProcessTask``:
 
 .. code-block:: java
 
     public class MyBundle implements BundleBuilder {
+      public static class ProcessTask implements Task {
+        @Override
+        public void execute(Context context, Client client) throws Exception {
+          var fetched = (String) client.getXCom("fetch");
+          // implement task logic
+          client.setXCom(fetched);
+        }
+      }
+
       @Override
       public Iterable<Dag> getDags() {
         var dag = new Dag("my_dag");

airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dag.py

@@ -71,6 +71,35 @@
 _LOG_FETCH_TIMEOUT = 60
 
 
+def _wait_for_task_log_record(
+    airflow_client: AirflowClient,
+    dag_id: str,
+    task_id: str,
+    run_id: str,
+    try_number: int,
+    match: Callable[[dict], bool],
+) -> tuple[dict | None, list[dict]]:
+    """Poll a task's logs until a record matching *match* appears.
+
+    Logs can lag behind the terminal task state, and earlier records arrive
+    before the one under test, so returning on any record would race. Keep
+    polling until the target record shows up or the deadline passes. Returns
+    the matching record (or ``None``) and the last batch of records seen for
+    diagnostics.
+    """
+    deadline = time.monotonic() + _LOG_FETCH_TIMEOUT
+    records: list[dict] = []
+    while True:
+        resp = airflow_client.get_task_logs(
+            dag_id=dag_id, run_id=run_id, task_id=task_id, try_number=try_number
+        )
+        records = [entry for entry in resp.get("content", []) if isinstance(entry, dict)]
+        record = next((r for r in records if match(r)), None)
+        if record is not None or time.monotonic() > deadline:
+            return record, records
+        time.sleep(3)
+
+
 class TestJavaSDKAnnotationExample:
     """Verify the annotation-based Java SDK example executes correctly."""
 
@@ -225,29 +254,6 @@ def test_load_retried_then_succeeded(self):
             f"try_number={load_ti.get('try_number')!r}, ti: {load_ti}"
         )
 
-    def _wait_for_transform_log_record(
-        self, run_id: str, try_number: int, match: Callable[[dict], bool]
-    ) -> tuple[dict | None, list[dict]]:
-        """Poll the ``transform`` task logs until a record matching *match* appears.
-
-        Logs can lag behind the terminal task state, and earlier records (e.g. the
-        first transform line) arrive before the one under test, so returning on any
-        record would race. Keep polling until the target record shows up or the
-        deadline passes. Returns the matching record (or ``None``) and the last
-        batch of records seen for diagnostics.
-        """
-        deadline = time.monotonic() + _LOG_FETCH_TIMEOUT
-        records: list[dict] = []
-        while True:
-            resp = self.airflow_client.get_task_logs(
-                dag_id="java_annotation_example", run_id=run_id, task_id="transform", try_number=try_number
-            )
-            records = [entry for entry in resp.get("content", []) if isinstance(entry, dict)]
-            record = next((r for r in records if match(r)), None)
-            if record is not None or time.monotonic() > deadline:
-                return record, records
-            time.sleep(3)
-
     def test_application_logs_preserve_their_level(self):
         """A Java task's SLF4J ``logger.info`` must reach the UI as INFO, not ERROR.
 
@@ -280,7 +286,10 @@ def test_application_logs_preserve_their_level(self):
         )
 
         # transform logs `logger.info("Got variable {}", variable)` -> "Got variable 123".
-        record, records = self._wait_for_transform_log_record(
+        record, records = _wait_for_task_log_record(
+            self.airflow_client,
+            "java_annotation_example",
+            "transform",
             run_id,
             transform_ti.get("try_number", 1),
             lambda r: str(r.get("event", "")).startswith("Got variable"),
@@ -294,6 +303,74 @@ def test_application_logs_preserve_their_level(self):
         )
 
 
+class TestJavaSDKUninstantiableExample:
+    """Verify a task class the runner cannot instantiate fails with an actionable log."""
+
+    airflow_client = AirflowClient()
+
+    @pytest.mark.parametrize(
+        ("task_id", "expected_task_class"),
+        [
+            (
+                "missing_no_arg_constructor",
+                "org.apache.airflow.example.UninstantiableExampleBuilder$MissingNoArgConstructor",
+            ),
+            (
+                "non_static_inner",
+                "org.apache.airflow.example.UninstantiableExampleBuilder$NonStaticInner",
+            ),
+        ],
+    )
+    def test_uninstantiable_task_fails_with_actionable_error(self, task_id: str, expected_task_class: str):
+        """A task class the runner cannot instantiate fails with a clear log message."""
+        resp = self.airflow_client.trigger_dag(
+            "java_uninstantiable_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_uninstantiable_example",
+            run_id=run_id,
+            timeout=_JAVA_TASK_TIMEOUT,
+        )
+
+        ti_resp = self.airflow_client.get_task_instances(dag_id="java_uninstantiable_example", run_id=run_id)
+        ti_map = {ti["task_id"]: ti for ti in ti_resp.get("task_instances", [])}
+        ti = ti_map.get(task_id, {})
+
+        assert ti.get("state") == "failed", (
+            f"Java {task_id!r} task should fail cleanly.\n"
+            f"  task state : {ti.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()} }"
+        )
+
+        record, records = _wait_for_task_log_record(
+            self.airflow_client,
+            "java_uninstantiable_example",
+            task_id,
+            run_id,
+            ti.get("try_number", 1),
+            lambda r: str(r.get("event", "")).startswith("Cannot instantiate task class"),
+        )
+        assert record is not None, (
+            f"{task_id!r} should emit a 'Cannot instantiate task class' record; "
+            f"events seen: {[r.get('event') for r in records]}"
+        )
+        assert record.get("event") == (
+            "Cannot instantiate task class. "
+            "A task class must be concrete and declare a public no-argument constructor"
+        ), f"instantiation error should carry the full actionable message; record: {record}"
+        assert str(record.get("level", "")).lower() == "error", (
+            f"instantiation error should be logged as ERROR, got {record.get('level')!r}; record: {record}"
+        )
+        assert record.get("taskClass") == expected_task_class, (
+            f"instantiation error should name the offending class {expected_task_class!r}, "
+            f"got {record.get('taskClass')!r}; record: {record}"
+        )
+
+
 # Each Scala task spins up its own local SparkSession; allow generous time for
 # three sequential JVM + Spark startups in a constrained CI container.
 _SPARK_TASK_TIMEOUT = 1200

java-sdk/example/src/java/org/apache/airflow/example/ExampleBundleBuilder.java

@@ -30,7 +30,8 @@ public Iterable<Dag> getDags() {
     return List.of(
         InterfaceExampleBuilder.build(),
         AnnotationExampleBuilder.build(),
-        XComCastingExampleBuilder.build());
+        XComCastingExampleBuilder.build(),
+        UninstantiableExampleBuilder.build());
   }
 
   public static void main(String[] args) {

java-sdk/example/src/java/org/apache/airflow/example/UninstantiableExampleBuilder.java

@@ -0,0 +1,52 @@
+/*
+ * 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 org.apache.airflow.sdk.*;
+import org.jetbrains.annotations.NotNull;
+
+public class UninstantiableExampleBuilder {
+  public static class MissingNoArgConstructor implements Task {
+    private final String marker;
+
+    public MissingNoArgConstructor(String marker) {
+      this.marker = marker;
+    }
+
+    public void execute(@NotNull Context context, Client client) {
+      throw new IllegalStateException(marker);
+    }
+  }
+
+  /**
+   * The not static only constructor implicitly takes the enclosing instance, so the
+   * runner's no-argument lookup fails even though no constructor is declared.
+   */
+  public class NonStaticInner implements Task {
+    public void execute(@NotNull Context context, Client client) {}
+  }
+
+  public static Dag build() {
+    var dag = new Dag("java_uninstantiable_example");
+    dag.addTask("missing_no_arg_constructor", MissingNoArgConstructor.class);
+    dag.addTask("non_static_inner", NonStaticInner.class);
+    return dag;
+  }
+}

java-sdk/example/src/resources/dags/java_examples.py

@@ -73,6 +73,14 @@ def produce_fraction(): ...
 def consume_float(): ...
 
 
+@task.stub(queue="java")
+def missing_no_arg_constructor(): ...
+
+
+@task.stub(queue="java")
+def non_static_inner(): ...
+
+
 @task()
 def python_task_2(transformed):
     print("python_task_2")
@@ -103,6 +111,13 @@ def java_xcom_casting_example():
     produce_fraction() >> consume_float()
 
 
+@dag(dag_id="java_uninstantiable_example")
+def java_uninstantiable_example():
+    missing_no_arg_constructor()
+    non_static_inner()
+
+
 java_interface_example()
 java_annotation_example()
 java_xcom_casting_example()
+java_uninstantiable_example()

java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt

@@ -364,4 +364,24 @@ class BuilderTest {
       "Cannot create task from vararg function t1",
     )
   }
+
+  @Test
+  @DisplayName("fail compilation when dag class is a non-static inner class")
+  fun failCompilationWhenDagClassIsNonStaticInner() {
+    val compilation =
+      compile(
+        """
+        package org.apache.airflow.example;
+        import org.apache.airflow.sdk.Builder;
+        public class TestExample {
+          @Builder.Dag
+          public class InnerDag { @Builder.Task(id = "foo") public void t1() {} }
+        }
+      """,
+      )
+    assertThat(compilation).failed()
+    assertThat(compilation).hadErrorContaining(
+      "an enclosing instance that contains org.apache.airflow.example.TestExample.InnerDag is required",
+    )
+  }
 }

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt

@@ -27,6 +27,7 @@ import org.apache.airflow.sdk.execution.comm.RetryTask
 import org.apache.airflow.sdk.execution.comm.StartupDetails
 import org.apache.airflow.sdk.execution.comm.SucceedTask
 import org.apache.airflow.sdk.execution.comm.TaskState
+import java.lang.reflect.InvocationTargetException
 import java.time.OffsetDateTime
 
 internal object TaskResult {
@@ -60,6 +61,8 @@ internal object TaskResult {
     it.endDate = endDate
     it.renderedMapIndex = renderedMapIndex
   }
+
+  fun failure(shouldRetry: Boolean) = if (shouldRetry) retry() else of(TaskState.State.FAILED)
 }
 
 internal object TaskRunner {
@@ -71,17 +74,29 @@ internal object TaskRunner {
     client: Client,
   ): Any {
     val task = bundle.dags[request.ti.dagId]?.tasks[request.ti.taskId] ?: return TaskResult.of(TaskState.State.REMOVED)
+    val instance =
+      try {
+        task.getDeclaredConstructor().newInstance()
+      } catch (e: InvocationTargetException) {
+        val cause = e.cause ?: e
+        logger.error(
+          "Task class constructor threw an exception",
+          mapOf("ti" to request.ti, "taskClass" to task.name, "error" to cause, "trace" to cause.stackTraceToString()),
+        )
+        return TaskResult.failure(request.tiContext.shouldRetry)
+      } catch (e: Throwable) {
+        logger.error(
+          "Cannot instantiate task class. A task class must be concrete and declare a public no-argument constructor",
+          mapOf("ti" to request.ti, "taskClass" to task.name, "error" to e, "trace" to e.stackTraceToString()),
+        )
+        return TaskResult.failure(request.tiContext.shouldRetry)
+      }
     return try {
-      task.getDeclaredConstructor().newInstance().execute(Context.from(request), client)
+      instance.execute(Context.from(request), client)
       TaskResult.success()
     } catch (e: Throwable) {
       logger.error("Error executing task", mapOf("ti" to request.ti, "error" to e, "trace" to e.stackTraceToString()))
-      e.printStackTrace()
-      if (request.tiContext.shouldRetry) {
-        TaskResult.retry()
-      } else {
-        TaskResult.of(TaskState.State.FAILED)
-      }
+      TaskResult.failure(request.tiContext.shouldRetry)
     }
   }
 }