From bd67a38be058e9728df4218a20802da0aa700057 Mon Sep 17 00:00:00 2001 From: PoAn Yang Date: Sat, 25 Jul 2026 10:35:56 +0900 Subject: [PATCH] Java SDK: Clearer error when a task class cannot be instantiated Signed-off-by: PoAn Yang --- .../language-sdks/java.rst | 22 ++- .../java_sdk_tests/test_java_sdk_dag.py | 125 ++++++++++++++---- .../airflow/example/ExampleBundleBuilder.java | 3 +- .../example/UninstantiableExampleBuilder.java | 52 ++++++++ .../src/resources/dags/java_examples.py | 15 +++ .../org/apache/airflow/sdk/BuilderTest.kt | 20 +++ .../org/apache/airflow/sdk/execution/Task.kt | 29 +++- .../apache/airflow/sdk/execution/TaskTest.kt | 78 +++++++++++ 8 files changed, 311 insertions(+), 33 deletions(-) create mode 100644 java-sdk/example/src/java/org/apache/airflow/example/UninstantiableExampleBuilder.java diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst index c8b73b4626a28..94960c37ce11e 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst +++ b/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 getDags() { var dag = new Dag("my_dag"); 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 e634ece89b7a0..1404fa25fd6c7 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 @@ -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 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 f63d6c4d74337..551f78e6d13a9 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 @@ -30,7 +30,8 @@ public Iterable getDags() { return List.of( InterfaceExampleBuilder.build(), AnnotationExampleBuilder.build(), - XComCastingExampleBuilder.build()); + XComCastingExampleBuilder.build(), + UninstantiableExampleBuilder.build()); } public static void main(String[] args) { diff --git a/java-sdk/example/src/java/org/apache/airflow/example/UninstantiableExampleBuilder.java b/java-sdk/example/src/java/org/apache/airflow/example/UninstantiableExampleBuilder.java new file mode 100644 index 0000000000000..5beed0aae53e6 --- /dev/null +++ b/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; + } +} diff --git a/java-sdk/example/src/resources/dags/java_examples.py b/java-sdk/example/src/resources/dags/java_examples.py index 5426911b0faee..cc596bb50e13e 100644 --- a/java-sdk/example/src/resources/dags/java_examples.py +++ b/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() 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 3e28e1b009afd..e417c969555c2 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 @@ -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", + ) + } } diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt index 60258dacc628d..5a7a1491671d8 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt +++ b/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) } } } diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt index 42a083b94aacb..d8bf0ca0b95f3 100644 --- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt @@ -35,6 +35,8 @@ import org.apache.airflow.sdk.execution.comm.TaskState import org.junit.jupiter.api.Assertions import org.junit.jupiter.api.DisplayName import org.junit.jupiter.api.Test +import java.io.ByteArrayOutputStream +import java.io.PrintStream import java.time.OffsetDateTime import java.util.UUID @@ -84,6 +86,61 @@ class TaskTest { Assertions.assertEquals(TaskState.State.FAILED, (result as TaskState).state) } + @Test + @DisplayName("Should not duplicate the stack trace to stderr when task fails") + fun shouldNotDuplicateStackTraceToStderrWhenTaskFails() { + val captured = ByteArrayOutputStream() + val original = System.err + System.setErr(PrintStream(captured)) + try { + runTask(bundleWith("failing", FailingTask::class.java), startupDetails(taskId = "failing"), noOpClient()) + } finally { + System.setErr(original) + } + + Assertions.assertEquals("", captured.toString()) + } + + @Test + @DisplayName("Should return failed and log an actionable message when task class has no public no-argument constructor") + fun shouldReturnFailedWhenTaskClassHasNoNoArgConstructor() { + LogSender.messages.clear() + val result = + runTask(bundleWith("uninstantiable", NoDefaultConstructorTask::class.java), startupDetails(taskId = "uninstantiable"), noOpClient()) + + Assertions.assertInstanceOf(TaskState::class.java, result) + Assertions.assertEquals(TaskState.State.FAILED, (result as TaskState).state) + val message = LogSender.messages.single { it.level == Level.ERROR } + Assertions.assertTrue(message.event.contains("public no-argument constructor")) { "unexpected event: ${message.event}" } + Assertions.assertEquals(NoDefaultConstructorTask::class.java.name, message.arguments["taskClass"]) + } + + @Test + @DisplayName("Should return failed and log the constructor failure when task class constructor throws") + fun shouldReturnFailedWhenTaskClassConstructorThrows() { + LogSender.messages.clear() + val result = + runTask(bundleWith("throwing", ThrowingConstructorTask::class.java), startupDetails(taskId = "throwing"), noOpClient()) + + Assertions.assertInstanceOf(TaskState::class.java, result) + Assertions.assertEquals(TaskState.State.FAILED, (result as TaskState).state) + val message = LogSender.messages.single { it.level == Level.ERROR } + Assertions.assertEquals("Task class constructor threw an exception", message.event) + Assertions.assertEquals(ThrowingConstructorTask::class.java.name, message.arguments["taskClass"]) + Assertions.assertInstanceOf(IllegalStateException::class.java, message.arguments["error"]) + Assertions.assertEquals("constructor boom", (message.arguments["error"] as Throwable).message) + } + + @Test + @DisplayName("Should return retry when task class cannot be instantiated and should_retry is true") + fun shouldReturnRetryWhenTaskClassCannotBeInstantiatedAndShouldRetryIsTrue() { + val details = startupDetails(taskId = "uninstantiable") + details.tiContext?.shouldRetry = true + val result = runTask(bundleWith("uninstantiable", NoDefaultConstructorTask::class.java), details, noOpClient()) + + Assertions.assertInstanceOf(RetryTask::class.java, result) + } + private fun bundleWith( taskId: String, taskClass: Class, @@ -171,4 +228,25 @@ class TaskTest { client: Client, ): Unit = throw NoClassDefFoundError("simulated") } + + class ThrowingConstructorTask : Task { + init { + throw IllegalStateException("constructor boom") + } + + override fun execute( + context: Context, + client: Client, + ) { + } + } + + class NoDefaultConstructorTask( + private val marker: String, + ) : Task { + override fun execute( + context: Context, + client: Client, + ): Unit = throw IllegalStateException(marker) + } }