code: def _run_job(
      self,
      cmd_to_run: str,
      env_vars: Dict[str, Any],
      working_dir: Optional[str] = None,
  ) -> int:
      self.counter += 1
      command_id = self.counter
      env = os.environ.copy()
      env["RAY_ADDRESS"] = self.cluster_manager.get_cluster_address()
      env.setdefault("ANYSCALE_HOST", str(ANYSCALE_HOST))
  
      full_cmd = (
          " ".join(f"{k}={v}" for k, v in env_vars.items())
          + " "
          + cmd_to_run
      )
      logger.info(
          f"Executing {cmd_to_run} with {env_vars} via ray job submit"
      )
  
      job_client = self._get_job_client()
  
      runtime_env = None
      if working_dir:
          runtime_env = {"working_dir": working_dir}
  
      job_id = job_client.submit_job(
          # Entrypoint shell command to execute
          entrypoint=full_cmd,
          runtime_env=runtime_env,
      )
      self.last_job_id = job_id
      self.job_id_pool[command_id] = job_id
      self.start_time[command_id] = time.time()
      return command_id
  
file_path: ray/release/ray_release/job_manager/job_manager.py
imports: import os
  import time
  from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.exception import CommandTimeout
  from ray_release.logger import logger
  from ray_release.util import ANYSCALE_HOST, exponential_backoff_retry
label_desc: Use image node_id 3 for calling the JobManager obj's underlying member method code with example usage: obj._run_job(cmd_to_run, env_vars, working_dir) and returns: command_id