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

Code before:
            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)
        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:

Review comment (suggestion):
First off, thank you @stephen-bracken for looking into this!

My question here is for clusters with limited resources and have Kueue enabled. The k8s Job will eventually spool up the pods but it will take a while if there is resource contention. In our case, it's a GPU cluster and many jobs reserve GPUs with potentially long runtimes. Could a configurable timeout instead of a retry number be used here? Or, don't have a retry number / timeout, and rely on DAG / task timeout instead (which is what my team preferred and is inherent in the solution I applied locally).

Thanks again!

Code after:
        :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


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`