Add Coordinator Layer and Java Coordinator

[jason810496]

Add Coordinator Layer and Java Coordinator

• Merge order: This is PR 2 of 3 — merge after Add Java SDK #65956

1. Add Java SDK #65956
2. Add Coordinator Layer and Java Coordinator #65958 (this PR)
3. Add CI, E2E Tests, and Pre-commit Hooks for Java SDK #65959

• Try it out: A combined PoC branch with all changes cherry-picked is available at [DON'T MERGE] Java SDK All #65960 for reviewers who want to test the full integration end-to-end.

Why

Airflow's DAG file processor and task runner only understand Python. To run DAGs and tasks authored in other languages (Java now, Go/Rust later), both the parsing pipeline and the execution pipeline need a language-agnostic extension point that delegates to an external runtime subprocess.

How

The Coordinator Abstraction

A new BaseCoordinator base class in the Task SDK (task-sdk/src/airflow/sdk/execution_time/coordinator.py) defines the extension point. Language providers subclass it and implement three methods:

Method	Purpose
can_handle_dag_file(bundle_name, path)	File Discovery (e.g., "is this a valid JAR that we can parse?")
dag_parsing_runtime_cmd(...)	Returns the subprocess command for DAG parsing
task_execution_runtime_cmd(...)	Returns the subprocess command for task execution

The base class owns the full subprocess lifecycle: TCP server creation, subprocess spawning, connection acceptance, and a selector-based byte-forwarding bridge between the Airflow supervisor (fd 0) and the language runtime (TCP socket). The shared I/O loop is extracted into selector_loop.py and reused by WatchedSubprocess.

Discovery and Routing

Providers register coordinators in provider.yaml under a new coordinators key. ProvidersManager (airflow-core) and ProvidersManagerTaskRuntime (task-sdk) both discover them:

• DAG Parsing: DagFileProcessorProcess._resolve_processor_target() iterates registered coordinators — the first whose can_handle_dag_file() returns True handles the file.
• Task Execution: task_runner._resolve_runtime_entrypoint() uses a two-step resolution: first it consults the [sdk] queue_to_sdk mapping (queue name to coordinator runtime name), then it falls back to matching DAG file extensions against registered coordinators.

Queue-Based Runtime Routing

Tasks are routed to non-Python runtimes via their queue assignment and a configuration mapping. Operators set queue="java-queue" (or any custom queue name), and the [sdk] queue_to_sdk config maps queue names to coordinator runtime names:

[sdk]
queue_to_sdk = {"java-queue": "java"}

This avoids adding new columns or API fields -- the existing queue field carries the routing signal from scheduling to execution, and the mapping is resolved at task execution time.

Java Provider

A new apache-airflow-providers-sdk-java provider implements JavaCoordinator:

• can_handle_dag_file: checks if the file is a JAR with valid Airflow Java SDK manifest attributes
• dag_parsing_runtime_cmd: constructs java -classpath <bundle>/* <MainClass> --comm=... --logs=...
• task_execution_runtime_cmd: handles both pure Java DAGs (JAR path) and Python stub DAGs (resolves bundle from [java] bundles_folder config)
• get_code_from_file: extracts embedded .java source from the JAR for Airflow UI display

What

Task SDK (task-sdk/)

• Add BaseCoordinator abstract base class with full subprocess bridge lifecycle
• Add selector_loop.py — shared selector-based I/O utilities, refactored out of supervisor.py
• Add _resolve_runtime_entrypoint() to task_runner.py with queue-based and file-extension-based dispatch
• Add QueueToCoordinatorMapper for resolving queue names to coordinators via [sdk] queue_to_sdk config
• Extract resolve_bundle() helper for reuse by both Python and coordinator paths
• Register coordinators discovery in ProvidersManagerTaskRuntime

Airflow Core (airflow-core/)

• Add [sdk] queue_to_sdk configuration option for queue-to-runtime mapping
• Extend DagFileProcessorProcess.start() with _resolve_processor_target() for coordinator delegation
• Extend DagFileProcessorManager to recognize runtime file extensions (e.g., .jar) and skip ZIP inspection for them
• Extend DagCode.get_code_from_file() to delegate to coordinator's get_code_from_file()
• Add coordinators extension point to provider.yaml.schema.json and provider_info.schema.json
• Register coordinators discovery in ProvidersManager

Java Provider (providers/sdk/java/)

• Add JavaCoordinator with DAG parsing, task execution, and code extraction
• Add BundleScanner for JAR manifest inspection and bundle resolution
• Add provider.yaml with coordinators registration and [java] bundles_folder config
• Add provider packaging (pyproject.toml, docs, LICENSE, NOTICE)
• Add java_sdk_setup.sh for Breeze development environment

---

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

• Yes, with help of Claude Code Ops 4.6 following the guidelines

Co-authored-by: Tzu-ping Chung uranusjr@gmail.com

[uranusjr]

Same for many other tests

[uranusjr]

I went through the AIP and it seems like this is up-to-date now. It still needs to wait until the AIP passes, but I think we can do reviews in the mean time.