code: def _run_job(
      self,
      cmd_to_run: str,
      env_vars: Dict[str, Any],
      working_dir: Optional[str] = None,
      upload_path: Optional[str] = None,
      pip: Optional[List[str]] = None,
  ) -> None:
      env = os.environ.copy()
      env.setdefault("ANYSCALE_HOST", str(ANYSCALE_HOST))
  
      logger.info(
          f"Executing {cmd_to_run} with {env_vars} via Anyscale job submit"
      )
  
      anyscale_client = self.sdk
  
      runtime_env = {
          "env_vars": env_vars,
          "pip": pip or [],
      }
      if working_dir:
          runtime_env["working_dir"] = working_dir
          if upload_path:
              runtime_env["upload_path"] = upload_path
  
      try:
          job_response = anyscale_client.create_job(
              CreateProductionJob(
                  name=self.cluster_manager.cluster_name,
                  description=f"Smoke test: {self.cluster_manager.smoke_test}",
                  project_id=self.cluster_manager.project_id,
                  config=dict(
                      entrypoint=cmd_to_run,
                      runtime_env=runtime_env,
                      build_id=self.cluster_manager.cluster_env_build_id,
                      compute_config_id=self.cluster_manager.cluster_compute_id,
                      max_retries=0,
                  ),
              ),
          )
      except Exception as e:
          raise JobStartupFailed(
              "Error starting job with name "
              f"{self.cluster_manager.cluster_name}: "
              f"{e}"
          ) from e
  
      self.last_job_result = job_response.result
      self.start_time = time.time()
  
      logger.info(f"Link to job: " f"{format_link(self.job_url)}")
      return
  
imports: import os
  import time
  import subprocess
  import tempfile
  from collections import deque
  from contextlib import contextmanager
  from typing import Any, Dict, Optional, Tuple, List
  from anyscale.sdk.anyscale_client.models import CreateProductionJob, HaJobStates
  from ray_release.anyscale_util import LAST_LOGS_LENGTH, get_cluster_name
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.exception import CommandTimeout, JobStartupTimeout, JobStartupFailed
  from ray_release.logger import logger
  from ray_release.signal_handling import register_handler, unregister_handler
  from ray_release.util import ANYSCALE_HOST, ERROR_LOG_PATTERNS, exponential_backoff_retry, anyscale_job_url, format_link
label_desc: Use image node_id 2 for calling the AnyscaleJobManager obj's underlying member method code with example usage: obj._run_job(cmd_to_run, env_vars, working_dir, upload_path, pip) without return types