raphaelauv · Oct 29, 2026
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
@@ -1748,6 +1748,8 @@ subprocess
 subprocesses
 subqueries
 subquery
+subresource
+subresources
 SubscriberClient
 subscriptionId
 substring

@@ -33,6 +33,12 @@ Changelog
   team's ``[kubernetes_executor]`` config. Like all team config, these don't fall back to the
   un-prefixed section. If your teams connected using the un-prefixed values, set them for each team.
 
+.. note::
+  Job operators and triggers now read the Job object instead of the ``jobs/status`` subresource, so they
+  need ``get`` on ``jobs`` (``batch`` API group) rather than on ``jobs/status``. If the ``ServiceAccount``
+  of your triggerer or workers was granted ``get`` on ``jobs/status`` but not on ``jobs``, grant it
+  ``get`` on ``jobs``.
+
 10.25.0
 .......
 

@@ -128,23 +128,24 @@ permissions:
      - apiGroups: ["batch"]
        resources: ["jobs"]
        verbs: ["create", "get", "list", "watch", "patch", "delete"]
-     - apiGroups: ["batch"]
-       resources: ["jobs/status"]
-       verbs: ["get", "watch"]
+
+The Job status is read from the Job object itself, so no permission on the
+``jobs/status`` subresource is needed.
 
 When a job operator waits for completion, streams pod logs, reads XCom, or
 cleans up discovered pods, it also needs the relevant pod permissions from
 the pod launch section.
 
 For deferrable job operators, the worker creates the Job before deferring and
-the triggerer polls Job status. If the triggerer uses in-cluster credentials,
-bind the triggerer's ``ServiceAccount`` to the Job status permissions. When
+the triggerer polls the Job. If the triggerer uses in-cluster credentials,
+grant the triggerer's ``ServiceAccount`` permission to get jobs. When
 XCom is enabled, the triggerer also needs permission to get pods and exec into
-the XCom sidecar container. The worker reads pod logs after the task resumes.
+the XCom sidecar container, and for ``KubernetesStartKueueJobOperator`` to list
+pods. The worker reads pod logs after the task resumes.
 
 ``KubernetesPatchJobOperator`` only needs permission to patch jobs.
-``KubernetesDeleteJobOperator`` reads Job status before deleting a Job, so it
-needs permission to get ``jobs/status`` and delete jobs.
+``KubernetesDeleteJobOperator`` reads the Job before deleting it, so it
+needs permission to get and delete jobs.
 
 Cleanup permissions
 -------------------

@@ -861,6 +861,23 @@ When the task is killed, ``on_kill`` deletes the Job (with foreground cascade).
 For discovered pods, deletion is controlled by ``on_kill_action``:
 ``delete_pod`` attempts direct pod deletion and ``keep_pod`` skips it.
 
+XCom of a Job
+^^^^^^^^^^^^^
+
+With ``wait_until_job_complete=True`` and ``do_xcom_push=True``, each pod of the Job writes its XCom to
+``/airflow/xcom/return.json``, as with ``KubernetesPodOperator``. The operator returns one value per pod,
+so the ``return_value`` XCom is a list. When the list has a single value, it is unwrapped and
+``return_value`` is that value, unless ``unwrap_single=False``. A pod that writes no XCom gives ``None``.
+
+The operator also pushes these XCom, whatever ``do_xcom_push`` is:
+
+* ``job_name`` and ``job_namespace``: the name and namespace of the Job, pushed when the Job is created.
+* ``job``: the Job as a dictionary, including its ``status`` (the finished Job when
+  ``wait_until_job_complete=True``).
+
+``KubernetesJobOperator`` reads the XCom of the pods it found when the Job started.
+For ``KubernetesStartKueueJobOperator``, see :ref:`howto/operator:KubernetesStartKueueJobOperator`.
+
 
 
 .. _howto/operator:KubernetesDeleteJobOperator:
@@ -930,6 +947,75 @@ you to start a Kueue job in a Kubernetes cluster
     :start-after: [START howto_operator_k8s_install_kueue]
     :end-before: [END howto_operator_k8s_install_kueue]
 
+Waiting for the Job
+^^^^^^^^^^^^^^^^^^^
+
+With ``wait_until_job_complete=True``, the operator waits on the Job itself rather than on its pods:
+the Job stays suspended without pods until Kueue admits it, and Kueue may preempt it (suspend it again
+and delete its running pods) then admit it again, any number of times. New pods, with new names, are
+created at each admission. Pods that already succeeded are not deleted by a preemption and still count
+for the Job. Once the Job is finished, logs are read from the pods the Job has at that time.
+
+XCom
+^^^^
+
+With ``do_xcom_push=True``, the XCom of each pod is read at the first poll after its base container ends,
+while the Job runs: until then, the XCom sidecar keeps the pod, and so the Job, from finishing. Once the Job is
+finished, only the XCom of the pods that are ``Succeeded`` at that time are kept: the XCom of pods
+deleted by a preemption and of failed pods are dropped. ``return_value`` is the list of those XCom,
+one per succeeded pod, ordered by pod name (see :ref:`howto/operator:kubernetesjoboperator` for the
+other XCom pushed and ``unwrap_single``).
+
+If the XCom of a succeeded pod could not be read, for example because the pod was deleted before the
+task finished, the task fails rather than returning an incomplete list. A pod that writes no XCom does
+not fail the task: its value is ``None``.
+
+Number of pods: ``completions`` and ``parallelism``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+The number of values in ``return_value`` is the number of pods that succeeded, set by the ``completions``
+and ``parallelism`` parameters of the Job:
+
+.. list-table::
+   :header-rows: 1
+
+   * - Settings
+     - Pods running at once
+     - The Job is finished when
+     - ``return_value``
+   * - Defaults: ``parallelism=1``, ``completions`` not set
+     - 1
+     - one pod succeeded
+     - the value of that pod
+   * - ``completions=N``
+     - up to ``parallelism``
+     - ``N`` pods succeeded
+     - a list of ``N`` values
+   * - ``parallelism=M`` (``M`` > 1), ``completions`` not set (work queue)
+     - up to ``M``
+     - one pod succeeded and all the others ended
+     - a list of one value per pod that succeeded, from 1 to ``M``
+
+Failed pods are retried up to ``backoff_limit`` and never add a value. ``completions`` can also come
+from ``full_job_spec`` or ``job_template_file`` when the parameter is not set.
+
+Pod names end with a random suffix, so the order of ``return_value`` does not follow the order in which
+pods ran. With ``completion_mode="Indexed"``, each pod gets its index, from ``0`` to ``completions - 1``,
+in the ``JOB_COMPLETION_INDEX`` environment variable, and pod names start with ``<job name>-<index>-``.
+As names are sorted as text (``10`` comes before ``2``), write the index into the XCom and sort the list
+on it in the downstream task when the order matters:
+
+.. code-block:: bash
+
+    echo "{\"index\": $JOB_COMPLETION_INDEX, \"rows\": 42}" > /airflow/xcom/return.json
+
+Kueue admits and preempts the Job as a whole: it reserves the quota for the pods the Job runs at once
+(``parallelism``, at most ``completions``), so a high ``parallelism`` may make the Job wait longer for
+admission.
+
+Reference
+^^^^^^^^^
+
 For further information, look at:
 
 * `Kubernetes Documentation <https://kubernetes.io/docs/home/>`__

@@ -176,6 +176,7 @@ triggers:
     python-modules:
       - airflow.providers.cncf.kubernetes.triggers.pod
       - airflow.providers.cncf.kubernetes.triggers.job
+      - airflow.providers.cncf.kubernetes.triggers.kueue
 
 
 secrets-backends:

@@ -73,6 +73,7 @@ def get_provider_info():
                 "python-modules": [
                     "airflow.providers.cncf.kubernetes.triggers.pod",
                     "airflow.providers.cncf.kubernetes.triggers.job",
+                    "airflow.providers.cncf.kubernetes.triggers.kueue",
                 ],
             }
         ],

@@ -633,13 +633,14 @@ def get_job_status(self, job_name: str, namespace: str) -> V1Job:
         """
         Get job with status of specified name and namespace.
 
+        Reads the Job object rather than the ``jobs/status`` subresource: both return the same ``status``,
+        but RBAC checks subresources separately, so this only needs ``get`` on ``jobs``.
+
         :param job_name: Name of Job to fetch.
         :param namespace: Namespace of the Job.
         :return: Job object
         """
-        return self.batch_v1_client.read_namespaced_job_status(
-            name=job_name, namespace=namespace, pretty=True
-        )
+        return self.batch_v1_client.read_namespaced_job(name=job_name, namespace=namespace, pretty=True)
 
     def wait_until_job_complete(self, job_name: str, namespace: str, job_poll_interval: float = 10) -> V1Job:
         """
@@ -1325,12 +1326,14 @@ async def get_job_status(self, name: str, namespace: str) -> V1Job:
         """
         Get job's status object.
 
+        Reads the Job object, see :meth:`KubernetesHook.get_job_status`.
+
         :param name: Name of the pod.
         :param namespace: Name of the pod's namespace.
         """
         async with self.get_conn() as connection:
             v1_api = async_client.BatchV1Api(connection)
-            job: V1Job = await v1_api.read_namespaced_job_status(
+            job: V1Job = await v1_api.read_namespaced_job(
                 name=name,
                 namespace=namespace,
             )

@@ -19,13 +19,26 @@
 from __future__ import annotations
 
 import json
+import time
 from collections.abc import Sequence
 from functools import cached_property
+from typing import TYPE_CHECKING, Any
 
 from kubernetes.utils import FailToCreateError
 
 from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
 from airflow.providers.cncf.kubernetes.operators.job import KubernetesJobOperator
+from airflow.providers.cncf.kubernetes.triggers.kueue import (
+    JOB_NAME_LABEL,
+    JOB_STATE_MESSAGES,
+    KubernetesKueueJobTrigger,
+    KueueJobState,
+    build_job_event,
+    collect_xcom_results,
+    get_job_state,
+    get_succeeded_pod_xcoms,
+)
+from airflow.providers.cncf.kubernetes.utils.pod_manager import XComRetrievalError
 from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
 from airflow.providers.common.compat.sdk import AirflowException
 
@@ -34,6 +47,9 @@
 else:
     from airflow.models import BaseOperator
 
+if TYPE_CHECKING:
+    from airflow.sdk import Context
+
 
 class KubernetesInstallKueueOperator(BaseOperator):
     """
@@ -105,6 +121,12 @@ class KubernetesStartKueueJobOperator(KubernetesJobOperator):
     """
     Executes a Kubernetes Job in Kueue.
 
+    With ``wait_until_job_complete=True``, the operator polls the Job itself until it is finished, instead
+    of waiting on the pods found when the Job started: Kueue keeps the Job suspended without pods until it
+    admits it, and may preempt it (suspending it again and deleting its pods) then admit it again any
+    number of times. The XCom of each pod is read while the Job runs, and kept only for the pods that are
+    ``Succeeded`` once the Job is finished. Logs are read from the pods the Job has once it is finished.
+
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
         :ref:`howto/operator:KubernetesStartKueueJobOperator`
@@ -133,3 +155,82 @@ def __init__(self, queue_name: str, *args, **kwargs) -> None:
             self.suspend = True
         self.labels.update({"kueue.x-k8s.io/queue-name": self.queue_name})
         self.annotations.update({"kueue.x-k8s.io/queue-name": self.queue_name})
+
+    def execute(self, context: Context):
+        if not self.wait_until_job_complete:
+            return super().execute(context)
+
+        self.name = self._set_name(self.name)
+        self.job_request_obj = self.build_job_request_obj(context)
+        self.job = self.create_job(job_request_obj=self.job_request_obj)  # must set `self.job` for `on_kill`
+
+        ti = context["ti"]
+        ti.xcom_push(key="job_name", value=self.job.metadata.name)
+        ti.xcom_push(key="job_namespace", value=self.job.metadata.namespace)
+
+        if self.deferrable:
+            self.defer(
+                trigger=KubernetesKueueJobTrigger(
+                    job_name=self.job.metadata.name,
+                    job_namespace=self.job.metadata.namespace,
+                    base_container_name=self.base_container_name,
+                    kubernetes_conn_id=self.kubernetes_conn_id,
+                    cluster_context=self.cluster_context,
+                    config_file=self.config_file,
+                    in_cluster=self.in_cluster,
+                    poll_interval=self.job_poll_interval,
+                    do_xcom_push=self.do_xcom_push,
+                ),
+                method_name="execute_complete",
+            )
+
+        return self.execute_complete(
+            context, self._wait_until_job_finished(self.job.metadata.name, self.job.metadata.namespace)
+        )
+
+    def _wait_until_job_finished(self, job_name: str, namespace: str) -> dict[str, Any]:
+        """Poll the Job until it is finished, and return the event ``KubernetesKueueJobTrigger`` yields."""
+        state: KueueJobState | None = None
+        xcom_results: dict[str, str] = {}
+        while True:
+            job = self.hook.get_job_status(job_name=job_name, namespace=namespace)
+            if (new_state := get_job_state(job)) != state:
+                state = new_state
+                self.log.info("The job '%s' %s", job_name, JOB_STATE_MESSAGES[state])
+            if state == KueueJobState.FINISHED:
+                return build_job_event(job, xcom_results)
+            if state == KueueJobState.RUNNING and self.do_xcom_push:
+                collect_xcom_results(
+                    self.client, self.pod_manager, job_name, namespace, self.base_container_name, xcom_results
+                )
+            time.sleep(self.job_poll_interval)
+
+    def execute_complete(self, context: Context, event: dict, **kwargs):
+        # The pods found before a preemption by Kueue were deleted: read the ones the Job has now
+        pods = sorted(
+            self.client.list_namespaced_pod(
+                namespace=event["namespace"], label_selector=f"{JOB_NAME_LABEL}={event['name']}"
+            ).items,
+            key=lambda pod: pod.metadata.name,
+        )
+        for pod in pods:
+            self.log_matching_pod(pod=pod, context=context)
+        xcom_result = get_succeeded_pod_xcoms(pods, event["xcom_result"])
+        result = super().execute_complete(
+            context,
+            {
+                **event,
+                "pod_names": [pod.metadata.name for pod in pods],
+                "pod_namespace": event["namespace"],
+                "xcom_result": xcom_result,
+            },
+            **kwargs,
+        )
+        # Only reached for a successful Job: super() raises for a failed one
+        succeeded = (event["job"].get("status") or {}).get("succeeded") or 0
+        if self.do_xcom_push and len(xcom_result) < succeeded:
+            raise XComRetrievalError(
+                f"Read the XCom of {len(xcom_result)} of the {succeeded} pods that succeeded in the job "
+                f"'{event['name']}': the other pods were deleted, or their XCom could not be read"
+            )
+        return result