[jason810496]

• closes: Handle custom xcom backends with go sdk (edge mode) #51067
• Depends on Vendor the Java SDK supervisor schema from this repository #75942

Changes since the third review, which leaves out the rebase so only the review response shows. 3ec4021 is the reviewed tree replayed onto the current base.

Why

Run the configured [core] xcom_backend at the supervisor, so TypeScript, Go and Java tasks store, read and purge XComs through it like Python tasks.

The backend runs in a helper process rather than in the supervisor itself:

• The supervisor runs in the long-lived worker process, which also forks Python task processes. Backend state left there, such as the fsspec filesystem that XComObjectStorageBackend caches, breaks Python tasks forked from it later (This class is not fork-safe).
• A backend call blocks on I/O. With the call in another process, the supervisor keeps heartbeating and can stop it on execution_timeout or when the server stops the task.

The helper starts on the first call the configured backend handles, so tasks on the default BaseXCom never start it.

How

• The supervisor of a language SDK task runs the backend on the runtime's XCom requests (SetXCom, GetXCom, GetXComSequenceItem, GetXComSequenceSlice, DeleteXCom) and on the XComs it clears when a try starts. With the default BaseXCom, requests go to the API server unchanged.
• The backend runs in a helper process, started per task the same way a Python task process is started (bare fork, or a new interpreter on macOS or with [core] execute_tasks_new_python_interpreter). So the long-lived worker process keeps none of the backend's state, such as a cached S3FileSystem that a Python task forked from it later could not use.
• Runtimes get what the default backend would give them: the value is decoded with serde before store_value, and a loaded value is encoded with serde again. skipmixin_key bypasses the backend for every task, since the scheduler reads it from the XCom row as it is. Operator extra-link keys (_link_*) bypass it for language SDK tasks only, so Python tasks keep the behaviour they have today.
• A backend call that fails fails the task. XCom that cannot be read or purged fails the task that is what a Python task does.
• The backend looks up Connections and Variables through the supervisor's API client.
• The supervisor services that process from the same request loop as the task, so a backend call blocks neither the loop nor its heartbeating, and a task whose backend hangs still stops at its execution_timeout.
• A backend call that fails — the backend could not run, or refused, the XCom — reaches the runtime as its own XCOM_BACKEND_ERROR rather than API_SERVER_ERROR, which would point a user at the API server instead of at their own backend.
• A new lang_sdk_xcom_object_storage e2e mode runs a Dag definition in each of TypeScript, Go and Java with XComObjectStorageBackend on LocalStack.

The backend needs no SDK change, since all three runtimes only talk to the supervisor. With this config, TypeScript, Go and Java tasks read and write offloaded XComs like Python tasks:

[core]
xcom_backend = airflow.providers.common.io.xcom.backend.XComObjectStorageBackend

Lifecycle

sequenceDiagram
    participant Runtime as Runtime (TS, Go, Java)
    participant Supervisor
    participant Backend as XCom backend process
    participant API as API server
    participant Storage as Object storage

    Note over Supervisor,Backend: The XCom backend process starts on the first call the backend handles,<br/>by fork, or by exec on macOS or with execute_tasks_new_python_interpreter
    Supervisor->>API: get and delete the XComs of the earlier try
    Supervisor->>Backend: purge(row)
    Backend->>Storage: delete object
    Runtime->>Supervisor: SetXCom(key, value)
    Supervisor->>Backend: store(coordinate, key, value)
    opt a Connection the backend's own secrets backends do not have
        Backend->>Supervisor: GetConnection
        Supervisor-->>Backend: Connection, read from the API server
    end
    Backend->>Storage: write value
    Backend-->>Supervisor: s3://... reference
    Supervisor->>API: set the XCom row to the reference
    Runtime->>Supervisor: GetXCom(task_id, key)
    Supervisor->>API: get the XCom row
    Supervisor->>Backend: load(row)
    Backend->>Storage: read value
    Backend-->>Supervisor: value
    Supervisor-->>Runtime: value
    Note over Supervisor,Backend: The supervisor answers the runtime once the call returns, and goes on heartbeating<br/>and enforcing execution_timeout while it runs
    Runtime->>Supervisor: exits
    Supervisor->>Backend: exit

Loading

---

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

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[ashb]

Having a whole separate process to handle xcom uplodads feels overkill. Why can't we do it from the coordinator process directly? The "why" in the PR desc doesn't address this

[jason810496]

Having a whole separate process to handle xcom uplodads feels overkill. Why can't we do it from the coordinator process directly? The "why" in the PR desc doesn't address this

Good point, I've added this to the Why in the description:

1. Fork safety. The supervisor runs inside the long-lived worker process (for example the Celery worker child), which also forks the Python task processes. A backend run there leaves its state in that process: XComObjectStorageBackend caches its fsspec filesystem (_STORE_CACHE in airflow.sdk.io.store), and a Python task forked from that process later fails with RuntimeError: This class is not fork-safe.
2. The supervisor stays in control. A backend call blocks on I/O. In the supervisor, a hung upload would block the loop that heartbeats and enforces execution_timeout, with no way to interrupt it. In a separate process, the supervisor keeps heartbeating during the call, and kills the helper on execution_timeout, when the server stops the task, or after too many failed heartbeats.

The helper is also lazy and per task: it starts on the first call the configured backend handles and serves all of that task's XCom calls, so tasks on the default BaseXCom never start it.

It is also not meant for XCom only. Any worker-side backend that runs in a Python task's process and keeps in-memory state should run in this helper for the same isolation, so the [workers] state_store_backend and the worker's secrets backends are next.

[Andrushika]

My 2 cents... I like the shape of keeping the backend out of the supervisor. But making this new XCom process a mini event loop (heartbeat, its own execution_timeout) feels kinda heavy... Is it possible to make the supervisor just hand the call to the helper and go back to its normal loop?

Then the only thing the XCom process does is run the backend call itself, and we can still monitor the XCom helper process from the same selector we use for the task runner socket. The supervisor can also get the result by listening on the XCom helper's socket, and reply to the task runner once it's done.

Maybe there's a reason this doesn't work -- curious what you ran into.

[jason810496]

Thanks for the review, I just updated this PR and #75942 which this PR depends on.

[uranusjr]

This needs changing?

[jason810496]

I'd like to it as-is. Explained in the comment below.

[uranusjr]

On 3.4 this isn't only about the XCom backend. The 3.4 bundle has no 2027-02-28, so resolve_version rejects it and every Java task fails, not just XCom backend calls. That's why main moved Go and TS to 3.5.0 with the same pin in #75387.

The Java pin can stay at 2026-10-30, since nothing here needs XCOM_BACKEND_ERROR on the Java side and #75942 decouples the pin from the vendored schema. If you'd rather bump the pin, min_airflow_version must be 3.5.0.