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

Signed-off-by: PoAn Yang <payang@apache.org>

Author: FrankYang0529

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

airflow-e2e-tests/docker/java.yml

@@ -19,15 +19,17 @@
 #
 # 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:
     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
+      - ./java-test-jars:/opt/airflow/java-test-jars:ro
+    command: celery worker -q java,scala,java-test,default

airflow-e2e-tests/java-test-bundle/.gitignore

@@ -0,0 +1,2 @@
+.gradle
+build/

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"
+}

airflow-e2e-tests/java-test-bundle/gradle.properties

@@ -0,0 +1,20 @@
+# 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.
+
+org.gradle.configuration-cache=true
+
+projectVersion=1.0.0-SNAPSHOT

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"

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

airflow-e2e-tests/java-test-bundle/src/resources/dags/java_test_dags.py

@@ -0,0 +1,38 @@
+# 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.
+"""Stub Dags for the Java runner-behaviour E2E fixtures in the java-test-bundle."""
+
+from __future__ import annotations
+
+from airflow.sdk import dag, task
+
+
+@task.stub(queue="java-test")
+def missing_no_arg_constructor(): ...
+
+
+@task.stub(queue="java-test")
+def non_static_inner(): ...
+
+
+@dag(dag_id="java_uninstantiable")
+def java_uninstantiable():
+    missing_no_arg_constructor()
+    non_static_inner()
+
+
+java_uninstantiable()

airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py

@@ -51,6 +51,9 @@
     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,25 +379,30 @@ 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
     # Main-Class would make JavaCoordinator's Main-Class discovery ambiguous, so
     # 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,19 +419,21 @@ 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")
     copyfile(
         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

airflow-e2e-tests/tests/airflow_e2e_tests/constants.py

@@ -77,6 +77,13 @@
 )
 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"