class_docstr: Watches for Kubernetes jobs.
class_docstr_tok: ["Watches","for","Kubernetes","jobs","."]
code: def process_status(
      self,
      pod_name: str,
      namespace: str,
      status: str,
      annotations: dict[str, str],
      resource_version: str,
      event: Any,
  ) -> None:
      pod = event["object"]
      annotations_string = annotations_for_logging_task_metadata(
          annotations
      )
      """Process status response."""
      if (
          event["type"] == "DELETED"
          and not pod.metadata.deletion_timestamp
      ):
          # This will happen only when the task pods are adopted by another executor.
          # So, there is no change in the pod state.
          # However, need to free the executor slot from the current executor.
          self.log.info(
              "Event: pod %s adopted, annotations: %s",
              pod_name,
              annotations_string,
          )
          self.watcher_queue.put(
              (
                  pod_name,
                  namespace,
                  ADOPTED,
                  annotations,
                  resource_version,
              )
          )
      elif status == "Pending":
          # deletion_timestamp is set by kube server when a graceful deletion is requested.
          # since kube server have received request to delete pod set TI state failed
          if (
              event["type"] == "DELETED"
              and pod.metadata.deletion_timestamp
          ):
              self.log.info(
                  "Event: Failed to start pod %s, annotations: %s",
                  pod_name,
                  annotations_string,
              )
              self.watcher_queue.put(
                  (
                      pod_name,
                      namespace,
                      TaskInstanceState.FAILED,
                      annotations,
                      resource_version,
                  )
              )
          else:
              self.log.debug(
                  "Event: %s Pending, annotations: %s",
                  pod_name,
                  annotations_string,
              )
      elif status == "Failed":
          self.log.error(
              "Event: %s Failed, annotations: %s",
              pod_name,
              annotations_string,
          )
          self.watcher_queue.put(
              (
                  pod_name,
                  namespace,
                  TaskInstanceState.FAILED,
                  annotations,
                  resource_version,
              )
          )
      elif status == "Succeeded":
          # We get multiple events once the pod hits a terminal state, and we only want to
          # send it along to the scheduler once.
          # If our event type is DELETED, we have the POD_EXECUTOR_DONE_KEY, or the pod has
          # a deletion timestamp, we've already seen the initial Succeeded event and sent it
          # along to the scheduler.
          if (
              event["type"] == "DELETED"
              or POD_EXECUTOR_DONE_KEY in pod.metadata.labels
              or pod.metadata.deletion_timestamp
          ):
              self.log.info(
                  "Skipping event for Succeeded pod %s - event for this pod already sent to executor",
                  pod_name,
              )
              return
          self.log.info(
              "Event: %s Succeeded, annotations: %s",
              pod_name,
              annotations_string,
          )
          self.watcher_queue.put(
              (pod_name, namespace, None, annotations, resource_version)
          )
      elif status == "Running":
          # deletion_timestamp is set by kube server when a graceful deletion is requested.
          # since kube server have received request to delete pod set TI state failed
          if (
              event["type"] == "DELETED"
              and pod.metadata.deletion_timestamp
          ):
              self.log.info(
                  "Event: Pod %s deleted before it could complete, annotations: %s",
                  pod_name,
                  annotations_string,
              )
              self.watcher_queue.put(
                  (
                      pod_name,
                      namespace,
                      TaskInstanceState.FAILED,
                      annotations,
                      resource_version,
                  )
              )
          else:
              self.log.info(
                  "Event: %s is Running, annotations: %s",
                  pod_name,
                  annotations_string,
              )
      else:
          self.log.warning(
              "Event: Invalid state: %s on pod: %s in namespace %s with annotations: %s with "
              "resource_version: %s",
              status,
              pod_name,
              namespace,
              annotations,
              resource_version,
          )
  
imports: from __future__ import annotations
  import contextlib
  import json
  import multiprocessing
  import time
  from queue import Empty, Queue
  from typing import TYPE_CHECKING, Any, Generic, TypeVar
  from kubernetes import client, watch
  from kubernetes.client.rest import ApiException
  from urllib3.exceptions import ReadTimeoutError
  from airflow.exceptions import AirflowException
  from airflow.providers.cncf.kubernetes.kube_client import get_kube_client
  from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import annotations_for_logging_task_metadata, annotations_to_key, create_pod_id
  from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 6 for calling the KubernetesJobWatcher obj's underlying member method code with example usage: obj.process_status(pod_name, namespace, status, annotations, resource_version, event) without return types