code: def test_job_finished_after_head_node_restart(
      ray_start_cluster_head_with_external_redis,
  ):
      cluster = ray_start_cluster_head_with_external_redis
      head_node = cluster.head_node
  
      # submit job
      client = JobSubmissionClient(head_node.address)
      submission_id = client.submit_job(
          entrypoint="python -c 'import ray; ray.init(); print(ray.cluster_resources()); \
              import time; time.sleep(1000)'"
      )
  
      def get_job_info(submission_id):
          gcs_client = GcsClient(cluster.address)
          all_job_info = gcs_client.get_all_job_info()
  
          return list(
              filter(
                  lambda job_info: "job_submission_id"
                  in job_info.config.metadata
                  and job_info.config.metadata["job_submission_id"]
                  == submission_id,
                  list(all_job_info.values()),
              )
          )
  
      def _check_job_running(submission_id: str) -> bool:
          job_infos = get_job_info(submission_id)
          if len(job_infos) == 0:
              return False
          job_info = job_infos[0].job_info
          return job_info.status == JobStatus.RUNNING
  
      # wait until job info is written in redis
      wait_for_condition(
          _check_job_running, submission_id=submission_id, timeout=10
      )
  
      # kill head node
      ray.shutdown()
      gcs_server_process = head_node.all_processes["gcs_server"][
          0
      ].process
      gcs_server_pid = gcs_server_process.pid
  
      cluster.remove_node(head_node)
  
      # Wait to prevent the gcs server process becoming zombie.
      gcs_server_process.wait()
      wait_for_pid_to_exit(gcs_server_pid, 1000)
  
      # restart head node
      cluster.add_node()
      ray.init(cluster.address)
  
      # verify if job is finished, which marked is_dead
      def _check_job_is_dead(submission_id: str) -> bool:
          job_infos = get_job_info(submission_id)
          if len(job_infos) == 0:
              return False
          job_info = job_infos[0]
          return job_info.is_dead
  
      wait_for_condition(
          _check_job_is_dead, submission_id=submission_id, timeout=10
      )
  
imports: import sys
  import os
  import threading
  from time import sleep
  import pytest
  import ray
  from ray._private.utils import get_or_create_event_loop
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  import ray._private.gcs_utils
  from ray._private import ray_constants
  from ray._private.test_utils import convert_actor_state, enable_external_redis, generate_system_config_map, wait_for_condition, wait_for_pid_to_exit, run_string_as_driver
  from ray.job_submission import JobSubmissionClient, JobStatus
  from ray._raylet import GcsClient
  import psutil
label_desc: Use image node_id 24 for calling a global function with example usage: test_job_finished_after_head_node_restart(ray_start_cluster_head_with_external_redis) and returns: list, unknown, job_info, False, False