Repository: apache/airflow
File: providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/job.py (Python)

Code before:
        return None

    def get_pods(
        self, pod_request_obj: k8s.V1Pod, context: Context, *, exclude_checked: bool = True
    ) -> Sequence[k8s.V1Pod]:
        """Return an already-running pods if exists."""
        label_selector = self._build_find_pod_label_selector(context, exclude_checked=exclude_checked)
        pod_list: Sequence[k8s.V1Pod] = []
        retry_number: int = 0

        while retry_number <= self.discover_pods_retry_number:
            if len(pod_list) == self.parallelism:
                break
            pod_list = self.client.list_namespaced_pod(
                namespace=pod_request_obj.metadata.namespace,
                label_selector=label_selector,
            ).items
            retry_number += 1

        if len(pod_list) == 0:
            raise AirflowException(f"No pods running with labels {label_selector}")

        for pod_instance in pod_list:
            self.log_matching_pod(pod=pod_instance, context=context)

        return pod_list


class KubernetesDeleteJobOperator(BaseOperator):
    """
    Delete a Kubernetes Job.

    .. seealso::
        For more information on how to use this operator, take a look at the guide:
        :ref:`howto/operator:KubernetesDeleteJobOperator`

    :param name: name of the Job.
    :param namespace: the namespace to run within kubernetes.
    :param kubernetes_conn_id: The :ref:`kubernetes connection id <howto/connection:kubernetes>`
        for the Kubernetes cluster.
    :param config_file: The path to the Kubernetes config file. (templated)
        If not specified, default value is ``~/.kube/config``
    :param in_cluster: run kubernetes client with in_cluster configuration.
    :param cluster_context: context that points to kubernetes cluster.
        Ignored when in_cluster is True. If None, current-context is used. (templated)
    :param delete_on_status: Condition for performing delete operation depending on the job status. Values:
        ``None`` - delete the job regardless of its status, "Complete" - delete only successfully completed
        jobs, "Failed" - delete only failed jobs. (default: ``None``)
    :param wait_for_completion: Whether to wait for the job to complete. (default: ``False``)
    :param poll_interval: Interval in seconds between polling the job status. Used when the `delete_on_status`
        parameter is set. (default: 10.0)

Review comment (bug):
The error message at line 465 states "No pods running with labels" but this might not be accurate. The pods could exist but not be in a running state yet (e.g., pending, initializing). Consider updating the error message to be more accurate, such as "Failed to find expected number of pods with labels" or "No pods found with labels" to avoid confusion.

Code after:
            return client_spec
        if client_spec and base_spec:
            client_spec.template.spec = PodGenerator.reconcile_specs(
                base_spec.template.spec, client_spec.template.spec
            )
            client_spec.template.metadata = PodGenerator.reconcile_metadata(
                base_spec.template.metadata, client_spec.template.metadata
            )
            return merge_objects(base_spec, client_spec)

        return None

    def get_pods(
        self, pod_request_obj: k8s.V1Pod, context: Context, *, exclude_checked: bool = True
    ) -> Sequence[k8s.V1Pod]:
        """Return an already-running pods if exists."""
        label_selector = self._build_find_pod_label_selector(context, exclude_checked=exclude_checked)
        pod_list: Sequence[k8s.V1Pod] = []
        retry_number: int = 0

        while retry_number <= self.discover_pods_retry_number:
            if len(pod_list) == self.parallelism:
                break
            pod_list = self.client.list_namespaced_pod(
                namespace=pod_request_obj.metadata.namespace,
                label_selector=label_selector,
            ).items
            retry_number += 1

        if len(pod_list) == 0:
            raise AirflowException(f"No pods running with labels {label_selector}")

        for pod_instance in pod_list:
            self.log_matching_pod(pod=pod_instance, context=context)

        return pod_list


class KubernetesDeleteJobOperator(BaseOperator):
    """
    Delete a Kubernetes Job.

    .. seealso::
        For more information on how to use this operator, take a look at the guide:
        :ref:`howto/operator:KubernetesDeleteJobOperator`

    :param name: name of the Job.
    :param namespace: the namespace to run within kubernetes.
    :param kubernetes_conn_id: The :ref:`kubernetes connection id <howto/connection:kubernetes>`
        for the Kubernetes cluster.
    :param config_file: The path to the Kubernetes config file. (templated)