Apply the configured XCom backend to language SDK task XComs

[jason810496]

• closes: Handle custom xcom backends with go sdk (edge mode) #51067

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.

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, as it does for Python tasks.
• The backend looks up Connections and Variables through the supervisor's API client.
• While a backend call runs, the supervisor keeps heartbeating, and stops the call when the task reaches its execution_timeout or the server stops the task.
• A Go task now fails when its return value cannot be stored, as TypeScript and Java tasks already do.
• A new lang_sdk_xcom_object_storage e2e mode runs a Dag definition in each of TypeScript, Go and Java with XComObjectStorageBackend on LocalStack.

No SDK change is needed, 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 R as Runtime (TS, Go, Java)
    participant S as Supervisor
    participant H as XCom backend process
    participant A as API server
    participant O as Object storage

    Note over S,H: H starts on the first call the backend handles,<br/>by fork, or by exec on macOS or with execute_tasks_new_python_interpreter
    S->>A: get and delete the XComs of the earlier try
    S->>H: purge(row)
    H->>O: delete object
    R->>S: SetXCom(key, value)
    S->>H: store(coordinate, key, value)
    H->>S: GetConnection
    S-->>H: Connection (from A)
    H->>O: write value
    H-->>S: s3://... reference
    S->>A: set the XCom row to the reference
    R->>S: GetXCom(task_id, key)
    S->>A: get the XCom row
    S->>H: load(row)
    H->>O: read value
    H-->>S: value
    S-->>R: value
    Note over S,H: While a call runs, S keeps heartbeating, and kills H<br/>when the task reaches execution_timeout or the server stops it
    R->>S: exits
    S->>H: 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