Apply the configured XCom backend to language SDK task XComs

.github/workflows/additional-prod-image-tests.yml

@@ -80,6 +80,10 @@ on:  # yamllint disable-line rule:truthy
         description: "Whether to run YAML Dag e2e tests (true/false)"
         required: true
         type: string
+      run-lang-sdk-xcom-e2e-tests:
+        description: "Whether to run language SDK XCom backend e2e tests (true/false)"
+        required: true
+        type: string
       providers-compatibility-tests-matrix:
         description: "JSON matrix of released Airflow versions for the OpenLineage e2e compat tests."
         required: true
@@ -400,6 +404,21 @@ jobs:
       (inputs.canary-run == 'true' && inputs.run-generic-prod-image-tests == 'true') ||
       inputs.run-yaml-dag-e2e-tests == 'true'
 
+  test-e2e-integration-tests-lang-sdk-xcom-object-storage:
+    name: "Language SDK XCom object storage backend tests with PROD image"
+    uses: ./.github/workflows/airflow-e2e-tests.yml
+    with:
+      workflow-name: "Language SDK XCom object storage backend e2e test"
+      runners: ${{ inputs.runners }}
+      platform: ${{ inputs.platform }}
+      default-python-version: "${{ inputs.default-python-version }}"
+      use-uv: ${{ inputs.use-uv }}
+      e2e_test_mode: "lang_sdk_xcom_object_storage"
+      java-sdk-version: "${{ inputs.java-sdk-version }}"
+    if: >-
+      (inputs.canary-run == 'true' && inputs.run-generic-prod-image-tests == 'true') ||
+      inputs.run-lang-sdk-xcom-e2e-tests == 'true'
+
   test-e2e-integration-tests-openlineage:
     name: "OpenLineage e2e tests with PROD image"
     uses: ./.github/workflows/airflow-e2e-tests.yml

.github/workflows/airflow-e2e-tests.yml

@@ -49,7 +49,7 @@ on:  # yamllint disable-line rule:truthy
         type: string
         required: true
       e2e_test_mode:
-        description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, ts_sdk, or yaml_dag"  # yamllint disable-line rule:line-length
+        description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, ts_sdk, yaml_dag, or lang_sdk_xcom_object_storage"  # yamllint disable-line rule:line-length
         type: string
         default: "basic"
       java-sdk-version:
@@ -84,7 +84,7 @@ on:  # yamllint disable-line rule:truthy
         type: string
         default: ""
       e2e_test_mode:
-        description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, ts_sdk, or yaml_dag"  # yamllint disable-line rule:line-length
+        description: "Test mode - basic, remote_log, remote_log_elasticsearch, remote_log_opensearch, xcom_object_storage, event_driven, java_sdk, go_sdk, ts_sdk, yaml_dag, or lang_sdk_xcom_object_storage"  # yamllint disable-line rule:line-length
         type: string
         default: "basic"
       java-sdk-version:
@@ -122,8 +122,10 @@ jobs:
           make-mnt-writeable-and-cleanup: true
         id: breeze
       # The java_sdk mode runs the Gradle builds, the go_sdk mode runs the Go
-      # build and the ts_sdk mode runs the pnpm build; every other e2e mode
-      # skips the toolchain steps below.
+      # build and the ts_sdk mode runs the pnpm build; the lang_sdk_xcom_object_storage
+      # mode runs all three, restoring but never saving the Gradle and Go caches of the
+      # java_sdk and go_sdk modes, since it builds only part of what they cache. Every
+      # other e2e mode skips the toolchain steps below.
       # LANG_SDK_NATIVE_TOOLCHAIN=true (the same switch the lang-SDK k8s job uses)
       # makes the e2e conftest build with the host toolchain provisioned here
       # instead of an ephemeral toolchain container, skipping the image pull; local
@@ -134,13 +136,13 @@ jobs:
       # (Spark alone is ~1GB) into a fresh ~/.gradle, whose ~/.gradle/jdks holds
       # arch-specific auto-provisioned JDKs (the example pins a Java 11 toolchain).
       - name: "Setup Java for the Java SDK build"
-        if: inputs.e2e_test_mode == 'java_sdk'
+        if: inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad  # v6.1.0
         with:
           distribution: 'temurin'
           java-version: ${{ inputs.java-sdk-version }}
       - name: "Restore Java SDK Gradle dependency cache"
-        if: inputs.e2e_test_mode == 'java_sdk'
+        if: inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         id: gradle-cache
         uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9  # v6.1.0
         with:
@@ -157,13 +159,17 @@ jobs:
       # separate version pin to keep in sync; cache: false because the explicit
       # restore/save pair below persists the module + build caches even on red runs.
       - name: "Setup Go for the Go SDK build"
-        if: inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'yaml_dag'
+        if: >-
+          inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'yaml_dag' ||
+          inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e  # v7.0.0
         with:
           go-version-file: go-sdk/go.mod
           cache: false
       - name: "Restore Go SDK module + build cache"
-        if: inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'yaml_dag'
+        if: >-
+          inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'yaml_dag' ||
+          inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         id: go-cache
         uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9  # v6.1.0
         with:
@@ -174,13 +180,13 @@ jobs:
           restore-keys: |
             e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}-
       - name: "Setup pnpm for the TypeScript SDK build"
-        if: inputs.e2e_test_mode == 'ts_sdk'
+        if: inputs.e2e_test_mode == 'ts_sdk' || inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         uses: pnpm/action-setup@ea17c68df8912ef543352723c149a84f56e3d413  # v6.1.0
         with:
           package_json_file: ts-sdk/package.json
           run_install: false
       - name: "Setup Node for the TypeScript SDK build"
-        if: inputs.e2e_test_mode == 'ts_sdk'
+        if: inputs.e2e_test_mode == 'ts_sdk' || inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage'
         uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1  # v7.1.0
         with:
           node-version: 22
@@ -191,7 +197,7 @@ jobs:
         env:
           DOCKER_IMAGE: "${{ inputs.docker-image-tag }}"
           E2E_TEST_MODE: "${{ inputs.e2e_test_mode }}"
-          LANG_SDK_NATIVE_TOOLCHAIN: "${{ (inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'ts_sdk' || inputs.e2e_test_mode == 'yaml_dag') && 'true' || 'false' }}"  # yamllint disable-line rule:line-length
+          LANG_SDK_NATIVE_TOOLCHAIN: "${{ (inputs.e2e_test_mode == 'java_sdk' || inputs.e2e_test_mode == 'go_sdk' || inputs.e2e_test_mode == 'ts_sdk' || inputs.e2e_test_mode == 'yaml_dag' || inputs.e2e_test_mode == 'lang_sdk_xcom_object_storage') && 'true' || 'false' }}"  # yamllint disable-line rule:line-length
       - name: "Save Java SDK Gradle dependency cache"
         # Saved even when the e2e tests fail: the Gradle warm-up is independent of
         # test outcome, and actions/cache's post step would drop it on every red run.

.github/workflows/ci-amd.yml

@@ -359,6 +359,7 @@ jobs:
       run-openlineage-e2e-compat-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-compat-tests }}
       run-ts-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ steps.selective-checks.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ steps.selective-checks.outputs.run-lang-sdk-xcom-e2e-tests }}
       run-system-tests: ${{ steps.selective-checks.outputs.run-system-tests }}
       run-task-sdk-tests: ${{ steps.selective-checks.outputs.run-task-sdk-tests }}
       run-task-sdk-integration-tests: ${{ steps.selective-checks.outputs.run-task-sdk-integration-tests }}
@@ -1185,6 +1186,7 @@ jobs:
         ${{ needs.build-info.outputs.providers-compatibility-tests-matrix }}
       run-ts-sdk-e2e-tests: ${{ needs.build-info.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ needs.build-info.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ needs.build-info.outputs.run-lang-sdk-xcom-e2e-tests }}
       use-uv: ${{ needs.build-info.outputs.use-uv }}
       run-ui-e2e-tests: ${{ needs.build-info.outputs.run-ui-e2e-tests }}
       run-airflow-ctl-integration-tests: ${{ needs.build-info.outputs.run-airflow-ctl-integration-tests }}

.github/workflows/ci-arm.yml

@@ -235,6 +235,7 @@ jobs:
       run-openlineage-e2e-compat-tests: ${{ steps.selective-checks.outputs.run-openlineage-e2e-compat-tests }}
       run-ts-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ steps.selective-checks.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ steps.selective-checks.outputs.run-lang-sdk-xcom-e2e-tests }}
       run-system-tests: ${{ steps.selective-checks.outputs.run-system-tests }}
       run-task-sdk-tests: ${{ steps.selective-checks.outputs.run-task-sdk-tests }}
       run-task-sdk-integration-tests: ${{ steps.selective-checks.outputs.run-task-sdk-integration-tests }}
@@ -1061,6 +1062,7 @@ jobs:
         ${{ needs.build-info.outputs.providers-compatibility-tests-matrix }}
       run-ts-sdk-e2e-tests: ${{ needs.build-info.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ needs.build-info.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ needs.build-info.outputs.run-lang-sdk-xcom-e2e-tests }}
       use-uv: ${{ needs.build-info.outputs.use-uv }}
       run-ui-e2e-tests: ${{ needs.build-info.outputs.run-ui-e2e-tests }}
       run-airflow-ctl-integration-tests: ${{ needs.build-info.outputs.run-airflow-ctl-integration-tests }}

.github/workflows/scheduled-legacy-prod-image-canary.yml

@@ -69,6 +69,7 @@ jobs:
       run-task-sdk-integration-tests: ${{ steps.selective-checks.outputs.run-task-sdk-integration-tests }}
       run-ts-sdk-e2e-tests: ${{ steps.selective-checks.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ steps.selective-checks.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ steps.selective-checks.outputs.run-lang-sdk-xcom-e2e-tests }}
       run-ui-e2e-tests: ${{ steps.selective-checks.outputs.run-ui-e2e-tests }}
       use-uv: ${{ steps.selective-checks.outputs.force-pip == 'true' && 'false' || 'true' }}
     steps:
@@ -237,6 +238,7 @@ jobs:
         ${{ needs.build-info.outputs.providers-compatibility-tests-matrix }}
       run-ts-sdk-e2e-tests: ${{ needs.build-info.outputs.run-ts-sdk-e2e-tests }}
       run-yaml-dag-e2e-tests: ${{ needs.build-info.outputs.run-yaml-dag-e2e-tests }}
+      run-lang-sdk-xcom-e2e-tests: ${{ needs.build-info.outputs.run-lang-sdk-xcom-e2e-tests }}
       use-uv: ${{ needs.build-info.outputs.use-uv }}
       run-ui-e2e-tests: ${{ needs.build-info.outputs.run-ui-e2e-tests }}
       run-airflow-ctl-integration-tests: ${{ needs.build-info.outputs.run-airflow-ctl-integration-tests }}

.pre-commit-config.yaml

@@ -364,17 +364,20 @@ repos:
         additional_dependencies: ['rich>=13.6.0']
         pass_filenames: false
       - id: sync-java-sdk-supervisor-schema
-        name: Sync Java SDK supervisor schema with configured version
+        name: Sync Java SDK supervisor schema with the distribution that owns it
         env:
           PYTHONPYCACHEPREFIX: .build/pycache
-        description: "Download the Supervisor Schema when Java SDK's bundled schema.json drifts"
+        description: "Copy task-sdk's supervisor schema when Java SDK's vendored schema.json drifts"
         entry: ./scripts/ci/prek/sync_java_sdk_supervisor_schema.py
         language: python
         pass_filenames: false
+        # Also watches the task-sdk source, like sync-go-sdk-schemas, so every schema change reaches the copy.
         files: >
           (?x)
+          ^task-sdk/src/airflow/sdk/execution_time/schema/schema\.json$|
           ^java-sdk/gradle\.properties$|
-          ^java-sdk/sdk/schema/schema\.json$
+          ^java-sdk/sdk/schema/schema\.json$|
+          ^scripts/ci/prek/sync_java_sdk_supervisor_schema\.py$
       - id: regenerate-java-sdk-verification-metadata
         name: Regenerate the Java SDK dependency verification metadata
         env:

airflow-core/docs/authoring-and-scheduling/language-sdks/index.rst

@@ -229,7 +229,7 @@ release that shipped it. See the legend below the tables for what each mark mean
      - 3.5.0
    * - Current Supervisor Schema Version
      - 2027-02-28
-     - 2026-10-30
+     - 2027-02-28
      - 2027-02-28
 
 .. list-table:: Common to both authoring modes
@@ -288,6 +288,10 @@ release that shipped it. See the legend below the tables for what each mark mean
      - ✓ 1.0.0
      - ✓ 1.0.0
      - ✓ 1.0.0
+   * - ``custom-xcom-backend`` (MAY)
+     - ✓ 1.0.0
+     - ✓ 1.0.0
+     - ✓ 1.0.0
    * - ``connection-read`` (MUST)
      - ✓ 1.0.0
      - ✓ 1.0.0

airflow-core/docs/core-concepts/xcoms.rst

@@ -123,6 +123,8 @@ A backend that only overrides ``serialize_value`` and ``deserialize_value`` keep
 with Dags that use task loops: every iteration of a task in a loop is a separate task instance, and only
 ``store_value`` can tell them apart.
 
+Tasks written with a language SDK, such as the TypeScript, Go or Java SDK, use the same backend. The worker runs the backend for those tasks, so install the backend and its provider on the worker, as for Python tasks.
+
 Verifying Custom XCom Backend usage in Containers
 -------------------------------------------------
 

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -344,11 +344,9 @@ def ti_run(
             )
 
         # Send the keys to the worker, which deletes those XComs. The server cannot delete them itself,
-        # because for a Python task the task runner also purges them from a custom XCom backend, and only
-        # workers can reach that backend. For a language SDK task (Java, Go, TypeScript), the supervisor
-        # deletes these XComs before the runtime starts the task and does not purge them, because those
-        # runtimes push their XComs straight to the API server, so the XCom values never pass through the
-        # backend.
+        # because the worker also purges them from a custom XCom backend, and only workers can reach that
+        # backend. For a language SDK task (Java, Go, TypeScript), the supervisor deletes and purges them
+        # before the runtime starts the task.
 
         # However, do not clear it for deferral
         xcom_keys: list[str] = []

airflow-e2e-tests/docker/lang-sdk-xcom-object-storage.yml

@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Docker Compose override for lang_sdk_xcom_object_storage E2E test mode, listed after
+# localstack.yml and ts.yml.
+#
+# One image runs the Dag definitions of all three language SDKs. It is the airflow-java-worker image
+# conftest builds (see java.yml), and ts.yml adds the node binary and the TypeScript bundle mount.
+# The Go and Java bundles sit in the Dags folder, which the Dag processor and the worker both mount.
+# The worker consumes the "typescript", "golang" and "java-native" queues that the Dags set.
+# USER and HOME are set for the statically linked Go bundle, as in go.yml.
+---
+services:
+  airflow-dag-processor:
+    image: airflow-java-worker
+    environment:
+      USER: airflow
+      HOME: /home/airflow
+
+  airflow-worker:
+    image: airflow-java-worker
+    environment:
+      USER: airflow
+      HOME: /home/airflow
+    command: celery worker -q typescript,golang,java-native,default