jason810496 opened a new pull request, #65956: URL: https://github.com/apache/airflow/pull/65956
# Add Java SDK for Apache Airflow > **Merge order:** This is PR **1 of 3** — merge this first. > 1. **`feature/java-sdk`** (this PR) — Java SDK > 2. `task-sdk/feature/coordinator-interface` — 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 To enable DAGs and tasks to be written entirely in Java (or mixed with Python via `@task.stub`), Airflow needs a Java SDK that speaks the Task SDK protocol — msgpack-framed communication over TCP for DAG parsing and task execution. ## How The Java SDK is a standalone Gradle project (`java-sdk/`) that implements the Airflow Task SDK protocol in Kotlin/Java. It communicates with the Airflow task runner through the same TCP comm/logs channels that the Python SDK uses, producing byte-compatible serialized DAG JSON. ### Architecture The SDK follows a client-server model where the JVM process connects to the coordinator's TCP servers: ``` Airflow Task Runner (Python) Java SDK Subprocess ──────────────────────────── ──────────────────── TCP comm server (127.0.0.1:random) ◄──── CoordinatorComm (msgpack framing) TCP logs server (127.0.0.1:random) ◄──── Logger (structured JSON) ``` - **DAG Parsing**: On `DagFileParseRequest`, the SDK calls `DagBundle.getDags()`, serializes to DagSerialization v3 format, and returns a `DagFileParsingResult`. - **Task Execution**: On `StartupDetails`, the SDK looks up the task, executes `Task.execute(Client)`, and sends `SucceedTask` / `TaskState`. During execution, the task can call `client.getConnection()`, `client.getVariable()`, `client.getXCom()`, `client.setXCom()` — each is a synchronous request/response over the comm channel. ### Serialization Compatibility Both SDKs share test cases in `test_dags.yaml`. A field-by-field comparison (`compare.py`) ensures the Java SDK produces structurally equivalent output to Python's `DagSerialization.serialize_dag()`. ## What - Add `java-sdk/` directory with the full SDK implementation: - **Public API**: `Task` interface, `Client`, `Dag`, `DagBundle`, `Bundle`, `Connection` - **Execution internals** (Kotlin): `CoordinatorComm` (msgpack TCP framing), `Serde` (DagSerialization v3 serializer), `TaskSdkFrames` (protocol message encoding/decoding), `TaskRunner`, `DagParser`, `Server`, `Supervisor`, `Logger` - **Build-time tooling**: `BundleInspector` (generates `airflow-metadata.yaml`), `BundleScanner` (discovers bundles from JAR manifests) - **Configuration**: `Config` (reads `airflow.cfg` for Java bundles folder, execution API URL, etc.) - Add `java-sdk/example/` with a complete `JavaExample` bundle demonstrating pure Java DAG authoring and `@task.stub` interop - Add `java-sdk/dags/stub_dag.py` demonstrating mixed Python + Java DAGs via `@task.stub(sdk="java")` - Add cross-language serialization validation infrastructure (`test_dags.yaml`, `serialize_python.py`, `compare.py`) - Add unit tests for `BundleScanner`, `CoordinatorComm`, `Serde`, `TaskRunner`, `DagParser`, `Config` --- ##### 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]
