Jason(Zhe-You) Liu
ebca0c8edc
* Java SDK: Register tasks as first-class TaskDef objects
`dag.addTask("extract", Extract.class)` stored tasks as a plain
`Map<String, Class<out Task>>`, which leaves nowhere to hang anything
else a task needs: dependency edges, task-level configuration, and
argument wiring all have to attach to a per-task object, and a map of
classes cannot carry them. Introducing that object now keeps those
follow-ups additive instead of forcing another break of the registration
API later.
The annotation surface keeps `Builder.Dag` / `Builder.Task`, and the
interface users implement keeps the `Task` name, so the definition
objects are `DagDef` and `TaskDef` -- a pairing that stays unambiguous
next to `Task` at a use site. The SDK is pre-1.0, so the old
string-keyed overload is removed outright rather than deprecated.
The KubernetesExecutor lang-SDK example adopts the new names directly rather
than pinning to the old ones: CI now builds those artifacts from the branch
under test (#71527), so the example no longer has to compile against whatever
the SDK on main happens to expose.
* Java SDK: Clarify task ownership validation
* Java SDK: Simplify manual task registration
Manual interface-based bundles are authored directly by users, so requiring every call to wrap the task class in TaskDef adds noise to the common case while the SDK can retain first-class task definitions internally.
Scala Spark example bundle
A Scala + Apache Spark bundle for the Java SDK, exercised by the java_sdk
end-to-end test. It shows a non-Java JVM language driving Spark from a
@task.stub task and routing Log4j 2 logs into Airflow via airflow-sdk-log4j2.
The scala_spark_example Dag chains three tasks, each running in its own JVM
with a local SparkSession and passing scalar results over XCom:
spark_extract- builds a DataFrame, pushes its row count.spark_transform- aggregates total revenue.spark_load- returns the persisted total.
Build
# From java-sdk/: publish the SDK to the local Maven repository first.
./gradlew publishToMavenLocal -PskipSigning=true
cd scala_spark_example
../gradlew bundle
fatJar is disabled, so build/bundle/ holds the bundle JAR plus every runtime
JAR (Spark included) — copy it into a Java coordinator's jars_root.
Running Spark under the Java SDK
Spark on Java 17 needs a set of --add-opens / --add-exports options that open
internal JDK modules (reflection, NIO, the off-heap cleaner, Kerberos, ...) to
Spark. spark-submit and the SparkSession builder inject these through Spark's
own launcher, but the Java SDK's JavaCoordinator starts the bundle JVM directly
and bypasses that launcher, so the coordinator has to pass them itself via
jvm_args:
{
"scala-jdk": {
"classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
"kwargs": {
"jars_root": ["/path/to/scala-jars"],
"main_class": "org.apache.airflow.example.ScalaSparkBundleBuilder",
"jvm_args": [
"-Xmx512m",
"-XX:+IgnoreUnrecognizedVMOptions",
"--add-opens=java.base/java.lang=ALL-UNNAMED",
"--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED",
"--add-opens=java.base/sun.nio.ch=ALL-UNNAMED"
]
}
}
}
The full list mirrors org.apache.spark.launcher.JavaModuleOptions for the Spark
version pinned in build.gradle. Spark may add or drop openings between releases,
so revisit it whenever you bump Spark. The end-to-end test keeps the authoritative,
complete copy in airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
(_SPARK_JAVA_MODULE_OPTIONS).