jason810496 opened a new pull request, #65958: URL: https://github.com/apache/airflow/pull/65958
# Add Coordinator Layer and Java Provider > **Merge order:** This is PR **2 of 3** — merge after `feature/java-sdk`. > 1. `feature/java-sdk` — Java SDK > 2. **`task-sdk/feature/coordinator-interface`** (this PR) — Language Coordinator Layer + Java Provider > 3. `ci/java-hardening` — CI, E2E Tests, and Pre-commit Hooks > > **Try it out:** A combined PoC branch with all changes cherry-picked is available at [`astronomer:airflow:feature/java-all`](https://github.com/apache/airflow/compare/main...astronomer:airflow:feature/java-all) 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()` matches `StartupDetails.ti.sdk` against coordinator `runtime_name` to dispatch to the correct runtime. ### The `sdk` Field A new nullable `sdk` column on `task_instance` carries the target SDK from scheduling to execution (`NULL` = Python). It flows through: `BaseOperator.sdk` → serialization → `task_instance.sdk` (DB) → Execution API `TaskInstance.sdk` → `StartupDetails` → task runner routing ### 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 `sdk` field to `BaseOperator`, `MappedOperator`, and the generated `TaskInstance` datamodel - Add `_resolve_runtime_entrypoint()` to `task_runner.py` for SDK-based dispatch - Extract `resolve_bundle()` helper for reuse by both Python and coordinator paths - Register `coordinators` discovery in `ProvidersManagerTaskRuntime` - Bump Execution API version to `2026-07-01` ### Airflow Core (`airflow-core/`) - Add `sdk` column to `task_instance` table (Alembic migration `0112`) - Add `sdk` field to the Execution API `TaskInstance` schema (Cadwyn version `v2026_07_01`) - 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` - Add `sdk` to `BaseOperator` serialization fields - Propagate `sdk` through `TaskInstance.insert_mapping()` and `refresh_from_task()` - Add `sdk` to the `ExecuteTask` workload ### 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? - [x] Yes, with help of Claude Code Ops 4.6 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) Co-authored-by: Tzu-ping Chung <[email protected]> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
