code: def _wait_job(self, command_id: int, timeout: int):
      from ray.job_submission import JobStatus  # noqa: F811
  
      start_time = time.monotonic()
      timeout_at = start_time + timeout
      next_status = start_time + 30
  
      while True:
          now = time.monotonic()
          if now >= timeout_at:
              raise CommandTimeout(
                  f"Cluster command timed out after {timeout} seconds."
              )
  
          if now >= next_status:
              logger.info(
                  f"... command still running ..."
                  f"({int(now - start_time)} seconds) ..."
              )
              next_status += 30
          status = self._get_job_status_with_retry(command_id)
          if status in {
              JobStatus.SUCCEEDED,
              JobStatus.STOPPED,
              JobStatus.FAILED,
          }:
              break
          time.sleep(1)
      status = self._get_job_status_with_retry(command_id)
      # TODO(sang): Propagate JobInfo.error_type
      if status == JobStatus.SUCCEEDED:
          retcode = 0
      else:
          retcode = -1
      duration = time.time() - self.start_time[command_id]
      return retcode, duration
  
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 5 for calling the JobManager obj's underlying member method code with example usage: obj._wait_job(command_id, timeout) and returns: retcode,  duration