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 (suggestion):
The loop logic has a potential issue where `pod_list` is assigned inside the loop after checking its length. This means that if the initial `pod_list` is empty (as initialized on line 452), the first iteration will always execute the list operation, which is correct. However, if fewer pods than expected are found, the loop continues retrying without any delay, which could result in rapid API calls. Consider adding a sleep between iterations to avoid overwhelming the Kubernetes API server.

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)