airflow-new-sdk

---
name: airflow-new-sdk
description: >
license: Apache-2.0
---

<!-- SPDX-License-Identifier: Apache-2.0
     https://www.apache.org/licenses/LICENSE-2.0 -->

# Implementing a new language SDK for Airflow

## Start here

**Read `contributing-docs/30_new_language_sdk.rst` first.** It is the
authoritative contributor guide for this topic — coordinator base class choices,
wire protocol spec, bundle footer format, and testing requirements. Everything
in this skill builds on top of it, not alongside it.

---

## Repository layout

Every new SDK needs two things. The coordinator (Python) goes here:

```
task-sdk/src/airflow/sdk/coordinators/<language>/
    __init__.py        # re-export + module docstring
    coordinator.py     # subclass of SubprocessCoordinator or BaseCoordinator
task-sdk/tests/coordinators/<language>/
    test_coordinator.py
task-sdk/tests/integration/coordinators/<language>/
    test_integration.py   # requires Breeze
```

The language SDK itself lives in a top-level `<language>-sdk/` directory (like
`java-sdk/` and `go-sdk/`). For native-executable languages using
`ExecutableCoordinator`, no coordinator code is needed at all.

---

## Choosing the right base class — quick guide

```
Does the runtime compile to a self-contained native executable?
  YES → Use ExecutableCoordinator (zero Python to write).
        Append an AFBNDL01 footer with a packer tool (see go-sdk reference).
  NO  →
    Does it start via a single shell command (node, ruby, dotnet, …)?
      YES → Subclass SubprocessCoordinator.
            Implement _build_execute_task_command only (see 30_new_language_sdk.rst).
      NO  →
        Subclass BaseCoordinator and implement execute_task from scratch.
        (Rare: gRPC daemons, shared memory, persistent processes.)
```

The full rationale, method signature, and socket lifecycle for each path are in
`30_new_language_sdk.rst`. Read that section before writing any code.

---

## Reference implementations to study

| What to study | Where |
|---|---|
| SubprocessCoordinator base class | `task-sdk/src/airflow/sdk/coordinators/_subprocess.py` |
| Java coordinator (SubprocessCoordinator subclass) | `task-sdk/src/airflow/sdk/coordinators/java/coordinator.py` |
| ExecutableCoordinator (native bundles) | `task-sdk/src/airflow/sdk/coordinators/executable/coordinator.py` |
| Wire protocol in Kotlin | `java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/` |
| Wire protocol in Go | `go-sdk/pkg/execution/` |
| AFBNDL01 footer (Go reference) | `go-sdk/internal/bundlefooter/`, `task-sdk/docs/executable-bundle-spec.rst` |
| All message types and field specs | `task-sdk/src/airflow/sdk/execution_time/schema/schema.json` |

The Java and Go implementations are the two production reference points. When
implementing a new SDK, read whichever matches the target language's runtime
model (JVM/interpreted → Java; native/compiled → Go).

---

## Logging

The **Logging** section of `30_new_language_sdk.rst` is the spec: the `--logs`
JSON record format, the level names, and the `AIRFLOW__LOGGING__*` environment
variables. Read it first. A few language-neutral details it leaves out:

- **Level values.** Levels follow Python's `logging` scale, so thresholds line
  up with the rest of Airflow: `CRITICAL=50`, `ERROR=40`, `WARNING=30`,
  `INFO=20`, `DEBUG=10`, `NOTSET=0`.
- **Parsing `NAMESPACE_LEVELS`.** Split the value on `[\s,]+`, then split each
  item on `=` into `(logger_name, level_name)`. Emit a record only when its
  level is `>=` the matching `logger_name` threshold, or the global threshold
  when no per-logger entry matches.
- **Don't drop late logs.** Connect the `--logs` socket early and keep it open
  until the `--comm` channel has finished; otherwise records emitted during
  teardown can be lost.
- **Extra config keys.** The runtime can't read Airflow's config, so if your SDK
  needs `[logging]` settings beyond the two above, propagate them as environment
  variables from your coordinator's `start`, the same way.

For how a given language wires its native logging frameworks into this channel,
read that SDK's source (e.g. `java-sdk/`) rather than reproducing it here.

### E2E test suite

Create two files mirroring `java_sdk_tests/` or `go_sdk_tests/`:

```
airflow-e2e-tests/tests/airflow_e2e_tests/<language>_sdk_tests/
    __init__.py
    test_<language>_sdk_dag.py
```

The test file should:

- Trigger the SDK's example Dag (the one added to `<language>-sdk/dags/` or
  equivalent) via `AirflowClient.trigger_dag`.
- Wait for the run to finish with `AirflowClient.wait_for_dag_run`.
- Assert that each SDK task instance reached `"success"`.
- Assert at least one XCom value — confirms the full round-trip from task
  return value through the supervisor to the XCom store.
- Assert structured logs where the SDK emits them (see the Go suite for an
  example of log-content assertions).

Run locally with:

```bash
E2E_TEST_MODE=<language>_sdk uv run --project airflow-e2e-tests pytest \
    tests/airflow_e2e_tests/<language>_sdk_tests/ -xvs
```

The Java (`java_sdk_tests/test_java_sdk_dag.py`) and Go
(`go_sdk_tests/test_go_sdk_dag.py`) suites are the reference implementations.


---

## PR checklist (items not covered by 30_new_language_sdk.rst)

The `30_new_language_sdk.rst` guide covers coordinator placement, wire protocol
implementation, and testing. These additional items belong in the same PR:

1. `task-sdk/src/airflow/sdk/coordinators/<language>/__init__.py` — short
   module docstring and `__all__` re-export of the coordinator class.
2. `airflow-core/docs/authoring-and-scheduling/language-sdks/<language>.rst` —
   user-facing doc following the structure of `java.rst` or `go.rst`.
3. `airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst` — add
   the new doc to the toctree.
4. `airflow-core/newsfragments/<PR>.feature.rst` — a new language is always
   user-visible; add a newsfragment.
5. CI wiring — check `dev/breeze/src/airflow_breeze/utils/selective_checks.py`
   to confirm `<language>-sdk/` changes trigger the right test group. Add if
   missing.
6. E2E tests — add a `<language>_sdk_tests/` suite under
   `airflow-e2e-tests/tests/airflow_e2e_tests/`. See below.