class_docstr: Launches PODS.
docstr: Launches the pod synchronously and waits for completion.
  
  :param image: image name
  :param code_path: path to the .py file for python and jar file for scala
  :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
  :return:
code: def start_spark_job(
      self, image=None, code_path=None, startup_timeout: int = 600
  ):
      """
      Launches the pod synchronously and waits for completion.
  
      :param image: image name
      :param code_path: path to the .py file for python and jar file for scala
      :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
      :return:
      """
      try:
          if image:
              self.body["spec"]["image"] = image
          if code_path:
              self.body["spec"]["mainApplicationFile"] = code_path
          self.log.debug("Spark Job Creation Request Submitted")
          self.spark_obj_spec = (
              self.custom_obj_api.create_namespaced_custom_object(
                  group=self.api_group,
                  version=self.api_version,
                  namespace=self.namespace,
                  plural=self.plural,
                  body=self.body,
              )
          )
          self.log.debug(
              "Spark Job Creation Response: %s", self.spark_obj_spec
          )
  
          # Wait for the driver pod to come alive
          self.pod_spec = k8s.V1Pod(
              metadata=k8s.V1ObjectMeta(
                  labels=self.spark_obj_spec["spec"]["driver"][
                      "labels"
                  ],
                  name=self.spark_obj_spec["metadata"]["name"]
                  + "-driver",
                  namespace=self.namespace,
              )
          )
          curr_time = dt.now()
          while self.spark_job_not_running(self.spark_obj_spec):
              self.log.warning(
                  "Spark job submitted but not yet started. job_id: %s",
                  self.spark_obj_spec["metadata"]["name"],
              )
              self.check_pod_start_failure()
              delta = dt.now() - curr_time
              if delta.total_seconds() >= startup_timeout:
                  pod_status = self.pod_manager.read_pod(
                      self.pod_spec
                  ).status.container_statuses
                  raise AirflowException(
                      f"Job took too long to start. pod status: {pod_status}"
                  )
              time.sleep(10)
      except Exception as e:
          self.log.exception(
              "Exception when attempting to create spark job"
          )
          raise e
  
      return self.pod_spec, self.spark_obj_spec
  
imports: from __future__ import annotations
  import time
  from copy import deepcopy
  from datetime import datetime
  from functools import cached_property
  import tenacity
  from kubernetes.client import CoreV1Api, CustomObjectsApi, models
  from kubernetes.client.rest import ApiException
  from airflow.exceptions import AirflowException
  from airflow.providers.cncf.kubernetes.resource_convert.configmap import convert_configmap, convert_configmap_to_volume
  from airflow.providers.cncf.kubernetes.resource_convert.env_variable import convert_env_vars
  from airflow.providers.cncf.kubernetes.resource_convert.secret import convert_image_pull_secrets, convert_secret
  from airflow.providers.cncf.kubernetes.utils.pod_manager import PodManager
  from airflow.utils.log.logging_mixin import LoggingMixin
label_desc: Use image node_id 4 for calling the CustomObjectLauncher obj's underlying member method code with example usage: obj.start_spark_job(image, code_path, startup_timeout) and returns: self,  self