ephraimbuddy commented on code in PR #35800:
URL: https://github.com/apache/airflow/pull/35800#discussion_r1404052934
##########
airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -642,39 +641,6 @@ def adopt_launched_task(
del tis_to_flush_by_key[ti_key]
self.running.add(ti_key)
- def _adopt_completed_pods(self, kube_client: client.CoreV1Api) -> None:
- """
- Patch completed pods so that the KubernetesJobWatcher can delete them.
-
- :param kube_client: kubernetes client for speaking to kube API
- """
- if TYPE_CHECKING:
- assert self.scheduler_job_id
-
- new_worker_id_label =
self._make_safe_label_value(self.scheduler_job_id)
- query_kwargs = {
- "field_selector": "status.phase=Succeeded",
- "label_selector": (
- "kubernetes_executor=True,"
-
f"airflow-worker!={new_worker_id_label},{POD_EXECUTOR_DONE_KEY}!=True"
- ),
- }
- pod_list = self._list_pods(query_kwargs)
- for pod in pod_list:
- self.log.info("Attempting to adopt pod %s", pod.metadata.name)
- from kubernetes.client.rest import ApiException
-
- try:
- kube_client.patch_namespaced_pod(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- body={"metadata": {"labels": {"airflow-worker":
new_worker_id_label}}},
- )
- except ApiException as e:
- self.log.info("Failed to adopt pod %s. Reason: %s",
pod.metadata.name, e)
Review Comment:
This should work and I feel it's better. 675 & 676
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
