code: def _handle_command_output(
      self,
      job_status_code: int,
      error: str,
      raise_on_timeout: bool = True,
  ):
      if job_status_code == -2:
          raise JobBrokenError(
              f"Job state is 'BROKEN' with error:\n{error}\n"
          )
  
      if job_status_code == -3:
          raise JobTerminatedError(
              "Job entered 'TERMINATED' state (it was terminated "
              "manually or Ray was stopped):"
              f"\n{error}\n"
          )
  
      if job_status_code == -4:
          raise JobTerminatedBeforeStartError(
              "Job entered 'TERMINATED' state before it started "
              "(most likely due to inability to provision required nodes; "
              "otherwise it was terminated manually or Ray was stopped):"
              f"\n{error}\n"
          )
  
      # First try to obtain the output.json from S3.
      # If that fails, try logs.
      try:
          output_json = self.fetch_output()
      except Exception:
          logger.exception("Exception when obtaining output from S3.")
          try:
              logs = self.get_last_logs()
              output_json = re.search(
                  r"### JSON \|([^\|]*)\| ###", logs
              )
              output_json = json.loads(output_json.group(1))
          except Exception:
              output_json = None
  
      workload_status_code = None
      if output_json:
          logger.info(f"Output: {output_json}")
          workload_status_code = output_json["return_code"]
          workload_time_taken = output_json["workload_time_taken"]
          prepare_return_codes = output_json["prepare_return_codes"]
          last_prepare_time_taken = output_json[
              "last_prepare_time_taken"
          ]
  
          # If we know results/metrics were not uploaded, we can fail fast
          # fetching later.
          self._results_uploaded = output_json["uploaded_results"]
          self._metrics_uploaded = output_json["uploaded_metrics"]
          self._artifact_uploaded = output_json["uploaded_artifact"]
  
          if prepare_return_codes and prepare_return_codes[-1] != 0:
              if prepare_return_codes[-1] == TIMEOUT_RETURN_CODE:
                  raise PrepareCommandTimeout(
                      "Prepare command timed out after "
                      f"{last_prepare_time_taken} seconds."
                  )
              raise PrepareCommandError(
                  f"Prepare command '{self.prepare_commands[-1]}' returned "
                  f"non-success status: {prepare_return_codes[-1]} with error:"
                  f"\n{error}\n"
              )
      else:
          raise JobNoLogsError("Could not obtain logs for the job.")
  
      if workload_status_code == TIMEOUT_RETURN_CODE:
          if not raise_on_timeout:
              # Expected - treat as success.
              return
  
          raise TestCommandTimeout(
              f"Command timed out after {workload_time_taken} seconds."
          )
  
      if workload_status_code is not None and workload_status_code != 0:
          raise TestCommandError(
              f"Command returned non-success status: {workload_status_code} with "
              f"error:\n{error}\n"
          )
  
      if job_status_code == -1:
          raise JobOutOfRetriesError(
              "Job returned non-success state: 'OUT_OF_RETRIES' "
              "(command has not been ran or no logs could have been obtained) "
              f"with error:\n{error}\n"
          )
  
imports: import json
  import os
  import re
  import tempfile
  import shlex
  from typing import TYPE_CHECKING, Any, Dict, Optional, List
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.command_runner.job_runner import JobRunner
  from ray_release.exception import TestCommandTimeout, TestCommandError, JobOutOfRetriesError, PrepareCommandError, PrepareCommandTimeout, JobBrokenError, FetchResultError, JobTerminatedBeforeStartError, JobNoLogsError, JobTerminatedError
  from ray_release.file_manager.job_file_manager import JobFileManager
  from ray_release.job_manager import AnyscaleJobManager
  from ray_release.logger import logger
  from ray_release.util import join_cloud_storage_paths, generate_tmp_cloud_storage_path, get_anyscale_sdk, S3_CLOUD_STORAGE
label_desc: Use image node_id 6 for calling the AnyscaleJobRunner obj's underlying member method code with example usage: obj._handle_command_output(job_status_code, error, raise_on_timeout) without return types