code: def test_job_head_choose_job_agent_E2E(
      ray_start_cluster_head_with_env_vars,
  ):
      cluster = ray_start_cluster_head_with_env_vars
      assert wait_until_server_available(cluster.webui_url) is True
      webui_url = cluster.webui_url
      webui_url = format_web_url(webui_url)
      client = JobSubmissionClient(webui_url)
  
      def submit_job_and_wait_finish():
          submission_id = client.submit_job(entrypoint="echo hello")
  
          wait_for_condition(
              _check_job_succeeded,
              client=client,
              job_id=submission_id,
              timeout=30,
          )
  
      head_http_port = DEFAULT_DASHBOARD_AGENT_LISTEN_PORT
      worker_1_http_port = 52366
      cluster.add_node(dashboard_agent_listen_port=worker_1_http_port)
      wait_for_condition(
          lambda: get_register_agents_number(webui_url) == 2, timeout=20
      )
      assert len(cluster.worker_nodes) == 1
      node_try_to_kill = list(cluster.worker_nodes)[0]
  
      def make_sure_worker_node_run_job(port):
          actors = ray.state.actors()
  
          def _kill_all_driver():
              for _, actor_info in actors.items():
                  if actor_info["State"] != "ALIVE":
                      continue
                  if actor_info["Name"].startswith(
                      "_ray_internal_job_actor"
                  ):
                      proc = psutil.Process(actor_info["Pid"])
                      try:
                          proc.kill()
                      except Exception:
                          pass
  
          try:
              for _, actor_info in actors.items():
                  if actor_info["State"] != "ALIVE":
                      continue
                  if actor_info["Name"].startswith(
                      "_ray_internal_job_actor"
                  ):
                      proc = psutil.Process(actor_info["Pid"])
                      parent_proc = proc.parent()
                      if f"--listen-port={port}" in " ".join(
                          parent_proc.cmdline()
                      ):
                          _kill_all_driver()
                          return True
          except Exception as ex:
              print("Got exception:", ex)
              raise
          client.submit_job(entrypoint="sleep 3600")
          return False
  
      # Make `list(cluster.worker_nodes)[0]` and head node called at least once
      wait_for_condition(
          lambda: make_sure_worker_node_run_job(worker_1_http_port),
          timeout=60,
      )
      wait_for_condition(
          lambda: make_sure_worker_node_run_job(head_http_port),
          timeout=60,
      )
  
      worker_2_http_port = 52367
      cluster.add_node(dashboard_agent_listen_port=worker_2_http_port)
      wait_for_condition(
          lambda: get_register_agents_number(webui_url) == 3, timeout=20
      )
  
      # The third `JobAgent` will not be called here.
      submit_job_and_wait_finish()
      submit_job_and_wait_finish()
      submit_job_and_wait_finish()
  
      def get_all_new_supervisor_actor_info(old_supervisor_actor):
          all_actors = ray.state.state.actor_table(None)
          res = dict()
          for actor_id, actor_info in all_actors.items():
              if actor_id in old_supervisor_actor:
                  continue
              if not actor_info["Name"].startswith(
                  "_ray_internal_job_actor"
              ):
                  continue
              res[actor_id] = actor_info
          return res
  
      old_supervisor_actor = set()
      new_supervisor_actor = get_all_new_supervisor_actor_info(
          old_supervisor_actor
      )
      new_owner_port = set()
      for actor_id, actor_info in new_supervisor_actor.items():
          old_supervisor_actor.add(actor_id)
          new_owner_port.add(actor_info["OwnerAddress"]["Port"])
  
      assert len(new_owner_port) == 2
      old_owner_port = new_owner_port
  
      node_try_to_kill.kill_raylet()
  
      # make sure the head updates the info of the dead node.
      wait_for_condition(
          lambda: get_register_agents_number(webui_url) == 2, timeout=20
      )
  
      # Make sure the third JobAgent will be called here.
      wait_for_condition(
          lambda: make_sure_worker_node_run_job(worker_2_http_port),
          timeout=60,
      )
  
      new_supervisor_actor = get_all_new_supervisor_actor_info(
          old_supervisor_actor
      )
      new_owner_port = set()
      for actor_id, actor_info in new_supervisor_actor.items():
          old_supervisor_actor.add(actor_id)
          new_owner_port.add(actor_info["OwnerAddress"]["Port"])
      assert len(new_owner_port) == 2
      assert len(old_owner_port - new_owner_port) == 1
      assert len(new_owner_port - old_owner_port) == 1
  
imports: from pathlib import Path
  import tempfile
  import time
  import os
  from ray.dashboard.modules.job.pydantic_models import JobType
  import requests
  import pytest
  import psutil
  import sys
  from typing import Dict, Optional, Tuple
  from unittest.mock import Mock, patch
  from ray._private.ray_constants import DEFAULT_DASHBOARD_AGENT_LISTEN_PORT
  from ray._private.test_utils import format_web_url, wait_for_condition, wait_until_server_available
  from ray.dashboard.modules.dashboard_sdk import ClusterInfo, DEFAULT_DASHBOARD_ADDRESS, parse_cluster_info
  from ray.dashboard.modules.job.sdk import JobSubmissionClient, JobStatus
  from ray.dashboard.tests.conftest import *
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR
  from ray.tests.conftest import _ray_start
  import ray
  import ray.experimental.internal_kv
  from ray.util.state import list_nodes
label_desc: Use image node_id 8 for calling a global function with example usage: test_job_head_choose_job_agent_E2E(ray_start_cluster_head_with_env_vars) and returns: False, res, True