class_docstr: Ray actor created by JobManager for each submitted job, responsible to
  setup runtime_env, execute given shell command in subprocess, update job
  status, persist job logs and manage subprocess group cleaning.
  
  One job supervisor actor maps to one subprocess, for one job_id.
  Job supervisor actor should fate share with subprocess it created.
docstr: Returns environment variables that should be set in the driver.
docstr_tok: ["Returns","environment","variables","that","should","be","set","in","the","driver","."]
code: def _get_driver_env_vars(
      self, resources_specified: bool
  ) -> Dict[str, str]:
      """Returns environment variables that should be set in the driver."""
      # RAY_ADDRESS may be the dashboard URL but not the gcs address,
      # so when the environment variable is not empty, we force set RAY_ADDRESS
      # to "auto" to avoid function `canonicalize_bootstrap_address_or_die` returning
      # the wrong GCS address.
      # TODO(Jialing He, Archit Kulkarni): Definition of Specification RAY_ADDRESS
      if ray_constants.RAY_ADDRESS_ENVIRONMENT_VARIABLE in os.environ:
          os.environ[
              ray_constants.RAY_ADDRESS_ENVIRONMENT_VARIABLE
          ] = "auto"
      ray_addr = (
          ray._private.services.canonicalize_bootstrap_address_or_die(
              "auto", ray.worker._global_node._ray_params.temp_dir
          )
      )
      assert ray_addr is not None
      return {
          # Set JobConfig for the child process (runtime_env, metadata).
          RAY_JOB_CONFIG_JSON_ENV_VAR: json.dumps(
              {
                  "runtime_env": self._get_driver_runtime_env(
                      resources_specified
                  ),
                  "metadata": self._metadata,
              }
          ),
          # Always set RAY_ADDRESS as find_bootstrap_address address for
          # job submission. In case of local development, prevent user from
          # re-using http://{address}:{dashboard_port} to interact with
          # jobs SDK.
          # TODO:(mwtian) Check why "auto" does not work in entrypoint script
          ray_constants.RAY_ADDRESS_ENVIRONMENT_VARIABLE: ray_addr,
          # Set PYTHONUNBUFFERED=1 to stream logs during the job instead of
          # only streaming them upon completion of the job.
          "PYTHONUNBUFFERED": "1",
      }
  
imports: import asyncio
  import copy
  import json
  import logging
  import os
  import psutil
  import random
  import signal
  import string
  import subprocess
  import sys
  import time
  import traceback
  from asyncio.tasks import FIRST_COMPLETED
  from collections import deque
  from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, SchedulingStrategyT
  import ray
  from ray._private.gcs_utils import GcsAioClient
  from ray._private.utils import run_background_task
  import ray._private.ray_constants
  from ray._private.runtime_env.constants import RAY_JOB_CONFIG_JSON_ENV_VAR
  from ray.actor import ActorHandle
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR, DEFAULT_JOB_START_TIMEOUT_SECONDS, RAY_JOB_START_TIMEOUT_SECONDS_ENV_VAR
  from ray.dashboard.modules.job.common import JOB_ID_METADATA_KEY, JOB_NAME_METADATA_KEY, JOB_ACTOR_NAME_TEMPLATE, JOB_LOGS_PATH_TEMPLATE, SUPERVISOR_ACTOR_RAY_NAMESPACE, JobInfo, JobInfoStorageClient
  from ray.dashboard.modules.job.utils import file_tail_iterator
  from ray.exceptions import ActorUnschedulableError, RuntimeEnvSetupError
  from ray.job_submission import JobStatus
  from ray._private.event.event_logger import get_event_logger
  from ray.core.generated.event_pb2 import Event
label_desc: Use image node_id 5 for calling the JobSupervisor obj's underlying member method code with example usage: obj._get_driver_env_vars(resources_specified) and returns: dict