Parse native Java Dags and show their source

[jason810496]

Stack (bottom to top): #74042, #74035, #74043, #74036, #74037, #73845, #73846, #73847

• Merge after Java SDK: Serialize native Dags to DagSerialization v3 #71190, which makes the Java runtime answer the Dag-parsing request, and Java SDK: Pack each native Dag's source file into the bundle JAR #74096, which packs the Dag sources.

Why

With this PR, JavaCoordinator parses the native Dags a bundle JAR declares and runs their tasks from that JAR, and the Code view shows their Java source.

What changes

• Which JARs. Once a JavaCoordinator is configured, each .jar whose manifest sets Main-Class (matching main_class if set) is parsed in every Dag bundle. An unreadable JAR is kept so its parse reports the error.
• Parse command. The task command, java -classpath <bundle JARs> <jvm_args> <Main-Class>, with the same schema version lookup. A JAR it cannot run gets an import error.
• Native tasks. A native Dag's task runs the JAR its Dag was parsed from, at the run's bundle version, on the JavaCoordinator its queue routes to. task_handler_bundle_name does not apply.
• Launch failures. A coordinator's execute_task raises the new TaskLaunchError when a native Dag's JAR cannot run, and the supervisor fails the task with the reason in its task log. A routing mistake, such as a queue routed to another kind of coordinator, fails without retries.
• Manifest reader. Tasks and parses now read manifests per the JAR specification (CR, LF or CRLF, folded lines, main section only). The old email parser broke a folded Main-Class.
• Dag source. get_source_code reads the sources.json index the Gradle plugin packs (Java SDK: Pack each native Dag's source file into the bundle JAR #74096), named by the Airflow-Java-SDK-Sources manifest attribute, and returns the entrypoint (Main-Class) source, up to 1 MiB per file, or a placeholder. The index also maps each Java Dag to its own file for a per-Dag lookup.

The Java runtime is unchanged here. The Java SDK layer above makes it answer the parse request, with a payload that passes validate_serialized_dag.

[sdk]
coordinators = {
  "java-native": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {"java_executable": "/usr/lib/jvm/java-17-openjdk/bin/java"}
  }
}
queue_to_coordinator = {"java-native": "java-native"}

Deviations and limitations

• Add ADR for Lang SDK Dag and mixed-language Task processing flow #71929 leaves get_source_code open for native Dags. This returns the source embedded as the pure Java Dags ADR (Move Lang-SDK ADRs from java-sdk/adr to airflow-core/adr/lang-sdk #70156) proposes, but under META-INF/airflow/sources/ instead of the JAR root to avoid fat JAR collisions, and without its airflow-metadata.yaml, since the parse runs the JVM.
• With several executable JARs in a bundle and no main_class, each is parsed, but a @task.stub task runs the first one in path order (Coordinator: detect ambiguous entrypoints and duplicate dag_ids at import time #71134).
• The Java SDK (Java SDK: Serialize native Dags to DagSerialization v3 #71190) still writes the stock values of max_active_tasks, max_active_runs, max_consecutive_failed_dag_runs, catchup and disable_bundle_versioning when a Dag leaves them unset, so a native Java Dag does not get the Airflow config values that Parse native Lang-SDK Dags in the Dag processor with their runtime #74035 fills in.
• Documented: tasks need queue set, no Dag in both Java and Python in one bundle, no cluster policies, airflow dags reserialize does not store a JAR's Dags, and airflow dags test, tasks test and tasks render refuse a native Java Dag.

---

Was generative AI tooling used to co-author this PR?

• Yes, with help of Claude Code Opus 5.5 following the guidelines

[jason810496]

Thanks for the review, I addressed them all in the latest commits.

Update: I will merge this with #74037 as soon as the CI pass there, so that we won't waste another CI run.