Read the k8s Java task handler bundle from S3 and test failure cases

.agents/skills/airflow-java-sdk/SKILL.md

@@ -50,8 +50,11 @@ subclasses must only import from `org.apache.airflow.sdk`; any import of
 
 ## Bundle composition and coordinator discovery
 
-A **bundle** is a directory of JAR files (typically `build/bundle/`) placed on the coordinator's
-`jars_root`. The coordinator scans the directory at task-dispatch time to find:
+A **bundle** is a directory of JAR files (typically `build/bundle/`) placed in the Dag bundle named
+by the coordinator's `task_handler_bundle_name` (the task's own Dag bundle when unset). The Dag
+processor lists the handler JARs in that Dag bundle and binds each stub task to the JAR that registers
+its handler. At task-dispatch time the coordinator runs only the JAR the task was bound to (or, for a
+Dag defined in Java, its own Dag file), and reads from it:
 
 1. **`Main-Class`** (standard JAR manifest attribute) — the fully-qualified class name of the
    entry point that the coordinator invokes with `java -classpath … <Main-Class> --comm … --logs …`.
@@ -65,16 +68,23 @@ A **bundle** is a directory of JAR files (typically `build/bundle/`) placed on t
    `runtimeClasspath` and copies it into the shadow JAR manifest. In thin-JAR mode (`fatJar =
    false`), the value stays in the `airflow-sdk` JAR deployed alongside the bundle JAR.
 
-The Python coordinator (`JavaCoordinator`) scans every JAR under `jars_root` with
-`_JarInfo.find()`, reads `META-INF/MANIFEST.MF` out of each ZIP, and collects `Main-Class` and
-`Airflow-Supervisor-Schema-Version` from whichever JARs carry them. The resolved schema version
-is then passed as the `schema_version` return value from `_build_execute_task_command`, which
-the base `SubprocessCoordinator` uses to negotiate the supervisor wire protocol.
+The Python coordinator (`JavaCoordinator`) reads `META-INF/MANIFEST.MF` out of that JAR in
+`_build_task_handler_command`, and takes `Main-Class` and `Airflow-Supervisor-Schema-Version` from
+it. A thin JAR that carries no schema version takes the first one found in the other JARs of the
+Dag bundle, in sorted path order. The resolved schema version is returned from
+`_build_task_handler_command`, and the base `SubprocessCoordinator` uses it to negotiate the
+supervisor wire protocol. The Dag processor probes a JAR with the same command.
 
 If `main_class` is set explicitly on the `JavaCoordinator` instance (via `[sdk] coordinators`
-kwargs), the scan uses it as a filter; otherwise the first JAR with a `Main-Class` attribute
-wins. Either way, `Airflow-Supervisor-Schema-Version` must be present in at least one JAR in
-`jars_root` or startup fails.
+kwargs), it runs instead of the manifest's `Main-Class`, for tasks and for the probe, so keep one
+handler JAR per Dag bundle then. The Java candidate's cache digest then covers `main_class`, so
+changing it makes the Dag processor probe again. Only a JAR whose manifest has `Airflow-Cache-Digest`
+is a handler JAR: the Gradle plugin writes it, and a Maven build must set a value that changes on
+every build (see java.rst). Either way, `Airflow-Supervisor-Schema-Version` must be present in
+at least one JAR in the Dag bundle. Without it the Dag processor cannot probe the JAR, so the stub
+Dag fails to import with the reason in its import error. A task fails before the JVM starts, with
+the reason in its task log, only when its worker's JARs differ from the Dag processor's. Every JAR
+in the Dag bundle goes on one classpath.
 
 ---
 
@@ -115,11 +125,12 @@ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
 
 ## Updating the Python coordinator
 
-`coordinator.py` extends `SubprocessCoordinator`. The only method subclasses must implement is
-`_build_execute_task_command`, which returns `(argv, schema_version)`. Look at the existing
-implementation for how `jars_root`, `java_executable`, `jvm_args`, and `main_class` are
-assembled into the command. Do not reach into the JVM process from Python beyond what this
-method provides.
+`coordinator.py` extends `SubprocessCoordinator`. The methods subclasses implement are
+`_read_task_handler_candidate`, which tells the coordinator's artifacts apart from the other files
+of a Dag bundle, and `_build_task_handler_command`, which returns `(argv, schema_version)` for the
+artifact at a path, for the probe and for a task alike. Look at the existing implementation for how
+the Dag bundle's JARs, `java_executable`, `jvm_args`, and `main_class` are assembled into the
+command. Do not reach into the JVM process from Python beyond what this method provides.
 
 ---
 

.agents/skills/airflow-new-sdk/SKILL.md

@@ -54,7 +54,8 @@ Does the runtime compile to a self-contained native executable?
   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).
+            Implement _build_task_handler_command and _read_task_handler_candidate
+            (see 30_new_language_sdk.rst).
       NO  →
         Subclass BaseCoordinator and implement execute_task from scratch.
         (Rare: gRPC daemons, shared memory, persistent processes.)

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

@@ -170,7 +170,7 @@ jobs:
           path: |
             ~/.cache/go-build
             ~/go/pkg/mod
-          key: e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('go-sdk/go.mod', 'go-sdk/go.sum') }}  # yamllint disable-line rule:line-length
+          key: e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('go-sdk/go.mod', 'go-sdk/go.sum', 'airflow-e2e-tests/go-test-bundle/go.sum') }}  # yamllint disable-line rule:line-length
           restore-keys: |
             e2e-go-sdk-go-v1-${{ runner.os }}-${{ runner.arch }}-
       - name: "Setup pnpm for the TypeScript SDK build"
@@ -192,6 +192,20 @@ jobs:
           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') && 'true' || 'false' }}"  # yamllint disable-line rule:line-length
+      # The Dag processor's probe of a real artifact has unit tests of its own that pack and probe the example
+      # of one Lang SDK. They need that SDK's toolchain, which is set up above for its mode, so they run here
+      # instead of in the core test jobs, and a mode that lost its toolchain fails them rather than skipping.
+      - name: "Probe the real ${{ inputs.e2e_test_mode }} task handler artifacts"
+        if: >-
+          (success() || failure())
+          && contains(fromJSON('["go_sdk", "ts_sdk", "java_sdk"]'), inputs.e2e_test_mode)
+        run: >
+          uv run --frozen --project airflow-core --with-editable shared/secrets_masker
+          pytest "airflow-core/tests/unit/dag_processing/test_task_handler_processor_${LANG_SDK}.py"
+          --color=yes -v -rs
+        env:
+          AIRFLOW_LANG_SDK_REAL_PROBE_TESTS: "1"
+          LANG_SDK: "${{ inputs.e2e_test_mode == 'go_sdk' && 'go' || inputs.e2e_test_mode == 'ts_sdk' && 'ts' || 'java' }}"  # 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/k8s-tests.yml

@@ -144,7 +144,7 @@ jobs:
   # them as workers under KubernetesExecutor. It is executor/naming/version-independent, so instead of
   # bolting it onto all six ``KubernetesExecutor``/``use-standard-naming == false`` system-test jobs above
   # (which ran it redundantly and added ~6 minutes each), it runs once here on a single default combo and
-  # executes only the lang-SDK test (``-k test_lang_sdk_combined_dag_succeeds``) - not the full suite.
+  # executes only the lang-SDK tests (``-k TestLangSdkCoordinatorExecutor``) - not the full suite.
   tests-kubernetes-lang-sdk:
     timeout-minutes: 60
     name: "K8S Lang-SDK:${{ inputs.lang-sdk-kubernetes-combo }}"
@@ -221,7 +221,7 @@ jobs:
       - name: "Run lang-SDK K8S test ${{ inputs.lang-sdk-kubernetes-combo }}"
         run: >-
           breeze k8s run-complete-tests --no-copy-local-sources
-          -- -k test_lang_sdk_combined_dag_succeeds
+          -- -k TestLangSdkCoordinatorExecutor
         env:
           EXECUTOR: "KubernetesExecutor"
           USE_STANDARD_NAMING: "false"

.pre-commit-config.yaml

@@ -390,6 +390,7 @@ repos:
           (?x)
           ^go-sdk/go\.mod$|
           ^kubernetes-tests/lang_sdk/go_example/go\.mod$|
+          ^airflow-e2e-tests/go-test-bundle/go\.mod$|
           ^\.github/workflows/ci-(arm|amd)\.yml$|
           ^airflow-e2e-tests/tests/airflow_e2e_tests/constants\.py$|
           ^dev/breeze/src/airflow_breeze/commands/kubernetes_commands\.py$|
@@ -412,20 +413,21 @@ repos:
         pass_filenames: false
         require_serial: true
       - id: check-go-example-mod-tidy
-        name: Check lang-SDK Go example module is tidy against the Go SDK
+        name: Check lang-SDK Go modules are tidy against the Go SDK
         entry: ./scripts/ci/prek/check_go_example_mod_tidy.py
         # golang so prek provisions the toolchain the check needs: static checks run on a
         # runner whose preinstalled toolchains are wiped to free disk space, and the SDK's
         # own `go mod tidy` hook gets its Go the same way.
         language: golang
-        # The example module keeps its own copy of the SDK's indirect requirements
-        # (it resolves the SDK through a `replace`), so a dependency moving in
-        # go-sdk/go.mod leaves it stale. Watching both modules' manifests is enough:
+        # The example modules keep their own copy of the SDK's indirect requirements
+        # (they resolve the SDK through a `replace`), so a dependency moving in
+        # go-sdk/go.mod leaves them stale. Watching all the manifests is enough:
         # any new requirement in the SDK necessarily lands in go-sdk/go.mod first.
         files: >
           (?x)
           ^go-sdk/go\.(mod|sum)$|
-          ^kubernetes-tests/lang_sdk/go_example/go\.(mod|sum)$
+          ^kubernetes-tests/lang_sdk/go_example/go\.(mod|sum)$|
+          ^airflow-e2e-tests/go-test-bundle/go\.(mod|sum)$
         pass_filenames: false
         require_serial: true
       - id: check-partition-mapper-defaults-in-sync

.rat-excludes

@@ -328,8 +328,9 @@ www-hash.txt
 # Vendored-in code
 /src/airflow/providers/google/_vendor/*
 
-# TypeScript bundle golden fixture: its bytes are digest-pinned, so no license header can be added
+# TypeScript bundle golden fixtures: their bytes are digest-pinned, so no license header can be added
 /ts-sdk/tests/cli/fixtures/bundle-v1.min.mjs
+/ts-sdk/tests/cli/fixtures/bundle-v1-with-task-handlers.min.mjs
 
 # Java SDK build outputs
 /java-sdk/bin/*