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 20229e47c99 Java SDK: Clearer error when a task class cannot be
instantiated (#69827)
20229e47c99 is described below
commit 20229e47c9966359967e269532916c1ba3ad2630
Author: PoAn Yang <[email protected]>
AuthorDate: Tue Aug 25 00:58:53 2026 +0900
Java SDK: Clearer error when a task class cannot be instantiated (#69827)
Signed-off-by: PoAn Yang <[email protected]>
---
.../language-sdks/java.rst | 23 +++-
airflow-e2e-tests/docker/java.yml | 12 +-
airflow-e2e-tests/java-test-bundle/.gitignore | 2 +
airflow-e2e-tests/java-test-bundle/build.gradle | 49 ++++++++
.../gradle.properties} | 19 +--
airflow-e2e-tests/java-test-bundle/settings.gradle | 33 ++++++
.../org/apache/airflow/e2e/TestBundleBuilder.java | 62 ++++++++++
.../src/resources/dags/java_test_dags.py} | 37 +++---
.../tests/airflow_e2e_tests/conftest.py | 45 ++++---
.../tests/airflow_e2e_tests/constants.py | 7 ++
.../java_sdk_tests/test_java_sdk_dag.py | 130 +++++++++++++++++----
dev/breeze/doc/ci/04_selective_checks.md | 7 ++
.../src/airflow_breeze/utils/selective_checks.py | 1 +
dev/breeze/tests/test_selective_checks.py | 9 ++
.../org/apache/airflow/sdk/execution/Task.kt | 37 ++++--
.../org/apache/airflow/sdk/execution/TaskTest.kt | 119 +++++++++++++++++++
16 files changed, 509 insertions(+), 83 deletions(-)
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 a3e27f03132..7fd346bf3f4 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,17 @@ Interface-based API
Implement the ``Task`` interface directly for full control over how tasks are
registered and how XComs are
read. Each task is registered as a ``TaskDef`` on a ``DagDef``.
+The runner creates a fresh instance of the task class through reflection for
every task-instance run,
+which puts four constraints on the class:
+
+* The task class itself must be ``public``.
+* It must be concrete: not abstract and not an interface.
+* It must declare a public no-argument constructor.
+* If nested inside another class, it must be a ``static`` nested 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 +229,21 @@ read. Each task is registered as a ``TaskDef`` on a
``DagDef``.
}
}
-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<DagDef> getDags() {
var dag = new DagDef("my_dag")
diff --git a/airflow-e2e-tests/docker/java.yml
b/airflow-e2e-tests/docker/java.yml
index d3609f4e8be..db4897f44c8 100644
--- a/airflow-e2e-tests/docker/java.yml
+++ b/airflow-e2e-tests/docker/java.yml
@@ -19,10 +19,11 @@
#
# Replaces the stock airflow-worker image with one that has a JRE installed
# (built by conftest._setup_java_sdk_integration via Dockerfile.java), mounts
-# the pre-built bundle JARs (the Java example under /opt/airflow/java-jars and
-# the Scala Spark example under /opt/airflow/scala-jars), and configures the
-# worker to consume the "java" and "scala" Celery queues where @task.stub tasks
-# are routed.
+# the pre-built bundle JARs (the Java example under /opt/airflow/java-jars, the
+# Scala Spark example under /opt/airflow/scala-jars, and the runner-behaviour
+# test fixtures under /opt/airflow/java-test-jars), and configures the worker
to
+# consume the "java", "scala", and "java-test" Celery queues where @task.stub
+# tasks are routed.
---
services:
airflow-worker:
@@ -30,4 +31,5 @@ services:
volumes:
- ./java-jars:/opt/airflow/java-jars:ro
- ./scala-jars:/opt/airflow/scala-jars:ro
- command: celery worker -q java,scala,default
+ - ./java-test-jars:/opt/airflow/java-test-jars:ro
+ command: celery worker -q java,scala,java-test,default
diff --git a/airflow-e2e-tests/java-test-bundle/.gitignore
b/airflow-e2e-tests/java-test-bundle/.gitignore
new file mode 100644
index 00000000000..7f6823bcc0f
--- /dev/null
+++ b/airflow-e2e-tests/java-test-bundle/.gitignore
@@ -0,0 +1,2 @@
+.gradle
+build/
diff --git a/airflow-e2e-tests/java-test-bundle/build.gradle
b/airflow-e2e-tests/java-test-bundle/build.gradle
new file mode 100644
index 00000000000..d826d2fe3f8
--- /dev/null
+++ b/airflow-e2e-tests/java-test-bundle/build.gradle
@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+plugins {
+ id("org.apache.airflow.sdk") version "${projectVersion}"
+}
+
+repositories {
+ mavenLocal()
+ mavenCentral()
+}
+
+dependencies {
+ implementation("org.apache.airflow:airflow-sdk:${projectVersion}")
+ implementation("org.apache.airflow:airflow-sdk-jpl:${projectVersion}")
+}
+
+java {
+ toolchain {
+ languageVersion.set(JavaLanguageVersion.of(11))
+ }
+ sourceCompatibility = JavaVersion.VERSION_11
+}
+
+sourceSets {
+ main {
+ java.srcDir("src/java")
+ }
+}
+
+airflowBundle {
+ mainClass = "org.apache.airflow.e2e.TestBundleBuilder"
+}
diff --git a/airflow-e2e-tests/docker/java.yml
b/airflow-e2e-tests/java-test-bundle/gradle.properties
similarity index 53%
copy from airflow-e2e-tests/docker/java.yml
copy to airflow-e2e-tests/java-test-bundle/gradle.properties
index d3609f4e8be..c3c94e0057f 100644
--- a/airflow-e2e-tests/docker/java.yml
+++ b/airflow-e2e-tests/java-test-bundle/gradle.properties
@@ -15,19 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-# Docker Compose override for java_sdk E2E test mode.
-#
-# Replaces the stock airflow-worker image with one that has a JRE installed
-# (built by conftest._setup_java_sdk_integration via Dockerfile.java), mounts
-# the pre-built bundle JARs (the Java example under /opt/airflow/java-jars and
-# the Scala Spark example under /opt/airflow/scala-jars), and configures the
-# worker to consume the "java" and "scala" Celery queues where @task.stub tasks
-# are routed.
----
-services:
- airflow-worker:
- image: airflow-java-worker
- volumes:
- - ./java-jars:/opt/airflow/java-jars:ro
- - ./scala-jars:/opt/airflow/scala-jars:ro
- command: celery worker -q java,scala,default
+org.gradle.configuration-cache=true
+
+projectVersion=1.0.0-SNAPSHOT
diff --git a/airflow-e2e-tests/java-test-bundle/settings.gradle
b/airflow-e2e-tests/java-test-bundle/settings.gradle
new file mode 100644
index 00000000000..1785aaf9168
--- /dev/null
+++ b/airflow-e2e-tests/java-test-bundle/settings.gradle
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+// Route the plugin lookup to the SDK build published to the local Maven
+// repository by conftest._setup_java_sdk_integration.
+pluginManagement {
+ repositories {
+ mavenLocal()
+ gradlePluginPortal()
+ }
+}
+
+plugins {
+ id("org.gradle.toolchains.foojay-resolver-convention") version "0.10.0"
+}
+
+rootProject.name = "airflow-e2e-java-test-bundle"
diff --git
a/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
b/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
new file mode 100644
index 00000000000..5986f9d57b0
--- /dev/null
+++
b/airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java
@@ -0,0 +1,62 @@
+/*
+ * 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.e2e;
+
+import java.util.List;
+import org.apache.airflow.sdk.*;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Bundle of deliberately broken task classes for the runner-behaviour E2E
tests.
+ */
+public class TestBundleBuilder implements BundleBuilder {
+ public static class MissingNoArgConstructor implements Task {
+ public MissingNoArgConstructor(String unused) {}
+
+ public void execute(@NotNull Context context, Client client) {
+ throw new IllegalStateException("should not be reachable");
+ }
+ }
+
+ /**
+ * A non-static nested class declares no constructor of its own, but the
implicit one
+ * takes the enclosing instance, so the runner's lookup for a no-argument
constructor
+ * fails.
+ */
+ public class NonStaticInner implements Task {
+ public void execute(@NotNull Context context, Client client) {
+ throw new IllegalStateException("should not be reachable");
+ }
+ }
+
+ @NotNull
+ @Override
+ public Iterable<DagDef> getDags() {
+ var dag = new DagDef("java_uninstantiable");
+ dag.addTask("missing_no_arg_constructor", MissingNoArgConstructor.class);
+ dag.addTask("non_static_inner", NonStaticInner.class);
+ return List.of(dag);
+ }
+
+ public static void main(String[] args) {
+ var bundle = new TestBundleBuilder().build();
+ Server.create(args).serve(bundle);
+ }
+}
diff --git a/airflow-e2e-tests/docker/java.yml
b/airflow-e2e-tests/java-test-bundle/src/resources/dags/java_test_dags.py
similarity index 53%
copy from airflow-e2e-tests/docker/java.yml
copy to airflow-e2e-tests/java-test-bundle/src/resources/dags/java_test_dags.py
index d3609f4e8be..09274b77cd9 100644
--- a/airflow-e2e-tests/docker/java.yml
+++ b/airflow-e2e-tests/java-test-bundle/src/resources/dags/java_test_dags.py
@@ -14,20 +14,25 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+"""Stub Dags for the Java runner-behaviour E2E fixtures in the
java-test-bundle."""
-# Docker Compose override for java_sdk E2E test mode.
-#
-# Replaces the stock airflow-worker image with one that has a JRE installed
-# (built by conftest._setup_java_sdk_integration via Dockerfile.java), mounts
-# the pre-built bundle JARs (the Java example under /opt/airflow/java-jars and
-# the Scala Spark example under /opt/airflow/scala-jars), and configures the
-# worker to consume the "java" and "scala" Celery queues where @task.stub tasks
-# are routed.
----
-services:
- airflow-worker:
- image: airflow-java-worker
- volumes:
- - ./java-jars:/opt/airflow/java-jars:ro
- - ./scala-jars:/opt/airflow/scala-jars:ro
- command: celery worker -q java,scala,default
+from __future__ import annotations
+
+from airflow.sdk import dag, task
+
+
[email protected](queue="java-test")
+def missing_no_arg_constructor(): ...
+
+
[email protected](queue="java-test")
+def non_static_inner(): ...
+
+
+@dag(dag_id="java_uninstantiable")
+def java_uninstantiable():
+ missing_no_arg_constructor()
+ non_static_inner()
+
+
+java_uninstantiable()
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 5b45974f053..2877a4dc7bb 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -51,6 +51,9 @@ from airflow_e2e_tests.constants import (
JAVA_SDK_EXAMPLE_LIBS_PATH,
JAVA_SDK_MAVEN_CACHE_PATH,
JAVA_SDK_ROOT_PATH,
+ JAVA_TEST_BUNDLE_DAGS_PATH,
+ JAVA_TEST_BUNDLE_LIBS_PATH,
+ JAVA_TEST_BUNDLE_ROOT_PATH,
KAFKA_DIR_PATH,
LANG_SDK_NATIVE_TOOLCHAIN,
LOCALSTACK_PATH,
@@ -376,13 +379,14 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
console.print("[yellow]Publishing Java SDK artifacts to local Maven
repository...")
_run_java_sdk_gradle(JAVA_SDK_ROOT_PATH, "publishToMavenLocal",
"-PskipSigning=true", native=native)
- # The example and scala_spark_example are independent Gradle builds that
both
- # consume the SDK artifact published above, so build them concurrently.
Sharing
- # a writable Gradle user home between concurrent builds is safe because
each
- # build can ping the other's lock-owner port over one shared loopback - the
- # host's own in native mode, --network=host in the container path (see the
- # helper's docstring); publishToMavenLocal has already unpacked the shared
- # wrapper distribution, so neither build races to fetch it.
+ # The example, scala_spark_example, and java-test-bundle are independent
+ # Gradle builds that all consume the SDK artifact published above, so build
+ # them concurrently. Sharing a writable Gradle user home between concurrent
+ # builds is safe because each build can ping the other's lock-owner port
over
+ # one shared loopback - the host's own in native mode, --network=host in
the
+ # container path (see the helper's docstring); publishToMavenLocal has
+ # already unpacked the shared wrapper distribution, so no build races to
+ # fetch it.
#
# The Gradle `bundle` task is a Copy that never prunes its destination, so
# JARs from an earlier build linger. A stale dependency JAR with its own
@@ -390,11 +394,15 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
# start each bundle from an empty directory.
rmtree(JAVA_SDK_EXAMPLE_LIBS_PATH, ignore_errors=True)
rmtree(SCALA_SPARK_EXAMPLE_LIBS_PATH, ignore_errors=True)
+ rmtree(JAVA_TEST_BUNDLE_LIBS_PATH, ignore_errors=True)
toolchain = "host toolchain" if native else "eclipse-temurin:17-jdk"
- console.print(f"[yellow]Building Java SDK and Scala Spark example bundles
concurrently ({toolchain})...")
+ console.print(
+ f"[yellow]Building Java SDK, Scala Spark, and test-fixture bundles
concurrently ({toolchain})..."
+ )
example_bundle_workdirs = [
JAVA_SDK_ROOT_PATH / "example",
JAVA_SDK_ROOT_PATH / "scala_spark_example",
+ JAVA_TEST_BUNDLE_ROOT_PATH,
]
with ThreadPoolExecutor(max_workers=len(example_bundle_workdirs)) as pool:
bundle_builds = [
@@ -411,6 +419,7 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
# expose them to the worker, and each JavaCoordinator globs its own dir.
copytree(JAVA_SDK_EXAMPLE_LIBS_PATH, tmp_dir / "java-jars")
copytree(SCALA_SPARK_EXAMPLE_LIBS_PATH, tmp_dir / "scala-jars")
+ copytree(JAVA_TEST_BUNDLE_LIBS_PATH, tmp_dir / "java-test-jars")
# Copy the Java SDK example Dag files so Airflow can discover them.
copyfile(JAVA_SDK_EXAMPLE_DAGS_PATH / "java_examples.py", tmp_dir / "dags"
/ "java_examples.py")
@@ -418,12 +427,13 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
SCALA_SPARK_EXAMPLE_DAGS_PATH / "scala_spark_examples.py",
tmp_dir / "dags" / "scala_spark_examples.py",
)
+ copyfile(JAVA_TEST_BUNDLE_DAGS_PATH / "java_test_dags.py", tmp_dir /
"dags" / "java_test_dags.py")
# Keep the bundle JARs out of the build context: Dockerfile.java only adds
a
# JRE and copies nothing from the context, so without this docker build
would
# tar and stream the bundles (hundreds of MB of Spark JARs) to the daemon
for
# nothing. The JARs reach the worker via the compose bind-mounts, not the
image.
- (tmp_dir / ".dockerignore").write_text("java-jars/\nscala-jars/\n")
+ (tmp_dir /
".dockerignore").write_text("java-jars/\nscala-jars/\njava-test-jars/\n")
# Build a local Docker image that extends DOCKER_IMAGE with a JRE.
# We do this explicitly so testcontainers' DockerCompose.start() does not
@@ -445,10 +455,11 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
check=True,
)
- # Two JavaCoordinators on the same worker image, one bundle per queue. The
- # scala-jdk entry pins main_class (Spark's large classpath makes Main-Class
- # discovery ambiguous) and carries Spark's Java 17 module openings, a small
- # driver heap, and a longer startup timeout for its large dependency
classpath.
+ # One JavaCoordinator per queue on the same worker image, each serving its
+ # own bundle. The scala-jdk entry pins main_class (Spark's large classpath
+ # makes Main-Class discovery ambiguous) and carries Spark's Java 17 module
+ # openings, a small driver heap, and a longer startup timeout for its large
+ # dependency classpath.
coordinator_config = json.dumps(
{
"java-jdk": {
@@ -464,9 +475,15 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
"task_startup_timeout": 60.0,
},
},
+ "java-test-jdk": {
+ "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
+ "kwargs": {"jars_root": ["/opt/airflow/java-test-jars"]},
+ },
}
)
- queue_to_coordinator = json.dumps({"java": "java-jdk", "scala":
"scala-jdk"})
+ queue_to_coordinator = json.dumps(
+ {"java": "java-jdk", "scala": "scala-jdk", "java-test":
"java-test-jdk"}
+ )
# Connection expected by the Java example bundle tasks. The JSON form
# covers all connection fields, in particular the port: wire integers
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
index cbba1dde33f..4a1a2f7be38 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
@@ -77,6 +77,13 @@ SCALA_SPARK_EXAMPLE_DAGS_PATH = (
)
SCALA_SPARK_EXAMPLE_LIBS_PATH = JAVA_SDK_ROOT_PATH / "scala_spark_example" /
"build" / "bundle"
+# Java test-fixture bundle paths (deliberately broken task classes for the
+# runner-behaviour E2E tests; a separate bundle with its own coordinator/queue
+# so they stay out of the user-facing example).
+JAVA_TEST_BUNDLE_ROOT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" /
"java-test-bundle"
+JAVA_TEST_BUNDLE_DAGS_PATH = JAVA_TEST_BUNDLE_ROOT_PATH / "src" / "resources"
/ "dags"
+JAVA_TEST_BUNDLE_LIBS_PATH = JAVA_TEST_BUNDLE_ROOT_PATH / "build" / "bundle"
+
# Go SDK E2E test paths
GO_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "go-sdk"
GO_SDK_DAGS_PATH = GO_SDK_ROOT_PATH / "dags"
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 e634ece89b7..c0ca18f0b37 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 @@ _JAVA_TASK_TIMEOUT = 600
_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 @@ class TestJavaSDKAnnotationExample:
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 @@ class TestJavaSDKAnnotationExample:
)
# 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,79 @@ class TestJavaSDKAnnotationExample:
)
+class TestJavaSDKUninstantiableTask:
+ """Verify a task class the runner cannot instantiate fails with an
actionable log.
+
+ The deliberately broken task classes live in the java-test-bundle fixture
+ project (served on the dedicated "java-test" queue), not in the user-facing
+ example bundle.
+ """
+
+ airflow_client = AirflowClient()
+
+ @pytest.mark.parametrize(
+ ("task_id", "expected_task_class"),
+ [
+ (
+ "missing_no_arg_constructor",
+
"org.apache.airflow.e2e.TestBundleBuilder$MissingNoArgConstructor",
+ ),
+ (
+ "non_static_inner",
+ "org.apache.airflow.e2e.TestBundleBuilder$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",
+ 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",
+ run_id=run_id,
+ timeout=_JAVA_TASK_TIMEOUT,
+ )
+
+ ti_resp =
self.airflow_client.get_task_instances(dag_id="java_uninstantiable",
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",
+ 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 public, 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/dev/breeze/doc/ci/04_selective_checks.md
b/dev/breeze/doc/ci/04_selective_checks.md
index 78848313242..be4fa4d4c82 100644
--- a/dev/breeze/doc/ci/04_selective_checks.md
+++ b/dev/breeze/doc/ci/04_selective_checks.md
@@ -441,6 +441,13 @@ together using `pytest-xdist` (pytest-xdist distributes
the tests among parallel
of affected providers (but not recursively - only direct dependencies are
added)
* if there are any changes to "common" provider code not belonging to any
provider (usually system tests
or tests), then tests for all Providers are run
+* `Java SDK E2E tests` (the `java_sdk` mode of the deployed-stack tests,
exposed as the
+ `run-java-sdk-e2e-tests` output) run when the Java SDK sources (`java-sdk/`,
excluding `.md`), the
+ Java test-fixture bundle (`airflow-e2e-tests/java-test-bundle/`), the Java
e2e suite or its Docker
+ files (`airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/`,
+ `airflow-e2e-tests/docker/java.yml`,
`airflow-e2e-tests/docker/Dockerfile.java`), or the Java
+ coordinator (`task-sdk/src/airflow/sdk/coordinators/java/`,
`_subprocess.py`) change. Like the
+ other deployed e2e suites, enabling them forces `PROD Image building`.
* `OpenLineage E2E tests` (the `openlineage` mode of the deployed-stack tests
under
`airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests`, exposed as the
`run-openlineage-e2e-tests` output) run when the `openlineage` or `common`
providers or the
diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
index 02c2e4bb929..d831e569227 100644
--- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
+++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
@@ -245,6 +245,7 @@ CI_FILE_GROUP_MATCHES: HashableDict[FileGroupForCi] =
HashableDict(
FileGroupForCi.JAVA_SDK_E2E_FILES: [
# `.md` excluded — doc-only edits do not affect the Gradle build.
r"^java-sdk/(?!.*\.md$).*",
+ r"^airflow-e2e-tests/java-test-bundle/.*",
r"^airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/.*",
r"^airflow-e2e-tests/docker/java\.yml$",
r"^airflow-e2e-tests/docker/Dockerfile\.java$",
diff --git a/dev/breeze/tests/test_selective_checks.py
b/dev/breeze/tests/test_selective_checks.py
index eb1c32eac64..447db93babf 100644
--- a/dev/breeze/tests/test_selective_checks.py
+++ b/dev/breeze/tests/test_selective_checks.py
@@ -1539,6 +1539,15 @@ def assert_outputs_are_printed(expected_outputs:
dict[str, str], stderr: str):
},
id="Skip ts-sdk docs build for a ts-sdk README-only change",
),
+ pytest.param(
+
("airflow-e2e-tests/java-test-bundle/src/java/org/apache/airflow/e2e/TestBundleBuilder.java",),
+ {
+ "run-java-sdk-tests": "false",
+ "run-java-sdk-e2e-tests": "true",
+ "prod-image-build": "true",
+ },
+ id="Run java e2e tests when the java test-fixture bundle changes",
+ ),
pytest.param(
("task-sdk/src/airflow/sdk/coordinators/java/coordinator.py",),
{
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 bddc5256f55..9f1a7537046 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
@@ -28,6 +28,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 {
@@ -61,6 +62,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 {
@@ -74,19 +77,39 @@ internal object TaskRunner {
val definition =
bundle.dags[request.ti.dagId]?.tasks[request.ti.taskId]?.definition
?: return TaskResult.of(TaskState.State.REMOVED)
+ val instance =
+ try {
+ definition.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 definition.name, "error" to
cause, "trace" to cause.stackTraceToString()),
+ )
+ // Retrying cannot help: instantiation fails the same way on every try.
+ return TaskResult.failure(shouldRetry = false)
+ } catch (e: ReflectiveOperationException) {
+ logger.error(
+ "Cannot instantiate task class. A task class must be public,
concrete, and declare a public no-argument constructor",
+ mapOf("ti" to request.ti, "taskClass" to definition.name, "error" to
e, "trace" to e.stackTraceToString()),
+ )
+ return TaskResult.failure(shouldRetry = false)
+ } catch (e: Throwable) {
+ // A valid class can still fail to initialize (static initializer,
linkage); a fresh JVM may succeed.
+ logger.error(
+ "Error initializing task class",
+ mapOf("ti" to request.ti, "taskClass" to definition.name, "error" to
e, "trace" to e.stackTraceToString()),
+ )
+ return TaskResult.failure(request.tiContext.shouldRetry)
+ }
return try {
-
definition.getDeclaredConstructor().newInstance().execute(Context.from(request),
client)
+ instance.execute(Context.from(request), client)
TaskResult.success()
} catch (e: CancellationException) {
throw e // Let coroutine cancellation propagate so the task coroutine
unwinds.
} 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 5d303a3dd81..1b1fd288ed7 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
@@ -36,6 +36,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
@@ -85,6 +87,88 @@ 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 failed and log an initialization failure when
the task class static initializer throws")
+ fun shouldReturnFailedWhenTaskClassStaticInitializerThrows() {
+ LogSender.messages.clear()
+ val result =
+ runTask(bundleWith("static_init", StaticInitFailureTask::class.java),
startupDetails(taskId = "static_init"), 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("Error initializing task class", message.event)
+ Assertions.assertEquals(StaticInitFailureTask::class.java.name,
message.arguments["taskClass"])
+ }
+
+ @Test
+ @DisplayName("Should return retry when the task class fails to initialize
and should_retry is true")
+ fun shouldReturnRetryWhenTaskClassFailsToInitializeAndShouldRetryIsTrue() {
+ val details = startupDetails(taskId = "static_init")
+ details.tiContext?.shouldRetry = true
+ val result = runTask(bundleWith("static_init",
StaticInitFailureTask::class.java), details, noOpClient())
+
+ Assertions.assertInstanceOf(RetryTask::class.java, result)
+ }
+
+ @Test
+ @DisplayName("Should return failed and never retry when task class cannot be
instantiated, even if should_retry is true")
+ fun
shouldReturnFailedEvenIfShouldRetryIsTrueWhenTaskClassCannotBeInstantiated() {
+ for (taskClass in listOf(NoDefaultConstructorTask::class.java,
ThrowingConstructorTask::class.java)) {
+ val details = startupDetails(taskId = "uninstantiable")
+ details.tiContext?.shouldRetry = true
+ val result = runTask(bundleWith("uninstantiable", taskClass), details,
noOpClient())
+
+ Assertions.assertInstanceOf(TaskState::class.java, result) { "unexpected
result for ${taskClass.simpleName}: $result" }
+ Assertions.assertEquals(TaskState.State.FAILED, (result as
TaskState).state)
+ }
+ }
+
private fun bundleWith(
taskId: String,
taskClass: Class<out Task>,
@@ -171,4 +255,39 @@ class TaskTest {
client: Client,
): Unit = throw NoClassDefFoundError("simulated")
}
+
+ class ThrowingConstructorTask : Task {
+ init {
+ throw IllegalStateException("constructor boom")
+ }
+
+ override fun execute(
+ context: Context,
+ client: Client,
+ ) {
+ }
+ }
+
+ class StaticInitFailureTask : Task {
+ companion object {
+ init {
+ throw IllegalStateException("static init boom")
+ }
+ }
+
+ override fun execute(
+ context: Context,
+ client: Client,
+ ) {
+ }
+ }
+
+ class NoDefaultConstructorTask(
+ unused: String,
+ ) : Task {
+ override fun execute(
+ context: Context,
+ client: Client,
+ ): Unit = throw IllegalStateException("should not be reachable")
+ }
}
