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

Code before:
        """
        if base_spec and not client_spec:
            return base_spec
        if not base_spec and client_spec:
            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
        parallelism = self.parallelism or 1  # Default to using single pod parallelism

        while len(pod_list) != parallelism or retry_number <= self.discover_pods_retry_number:
            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.

Review comment (suggestion):
The while loop condition uses `or` but should use `and`. With the current logic, the loop will continue as long as either `len(pod_list) != parallelism` OR `retry_number <= self.discover_pods_retry_number` is true. This means:
1. If pods are found matching the parallelism count, the loop continues as long as `retry_number <= self.discover_pods_retry_number`
2. If the retry number exceeds the limit but pods don't match, it still continues

The correct logic should be: `while len(pod_list) != parallelism and retry_number <= self.discover_pods_retry_number:` to retry finding pods until either the expected count is found OR the retry limit is reached.
```suggestion
        while len(pod_list) != parallelism and retry_number <= self.discover_pods_retry_number:
```

Code after:
    @staticmethod
    def reconcile_job_specs(
        base_spec: k8s.V1JobSpec | None, client_spec: k8s.V1JobSpec | None
    ) -> k8s.V1JobSpec | None:
        """
        Merge Kubernetes JobSpec objects.

        :param base_spec: has the base attributes which are overwritten if they exist
            in the client_spec and remain if they do not exist in the client_spec
        :param client_spec: the spec that the client wants to create.
        :return: the merged specs
        """
        if base_spec and not client_spec:
            return base_spec
        if not base_spec and client_spec:
            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