code: def test_fault_tolerance_job_failed(shutdown_only):
      sys_config = _SYSTEM_CONFIG.copy()
      config = {
          "gcs_mark_task_failed_on_job_done_delay_ms": 1000,
          # make worker failure not trigger task failure
          "gcs_mark_task_failed_on_worker_dead_delay_ms": 30000,
      }
      sys_config.update(config)
      ray.init(num_cpus=8, _system_config=sys_config)
      script = """
  import ray
  import time
  
  ray.init("auto")
  NUM_CHILD = 2
  
  @ray.remote
  def grandchild():
      time.sleep(999)
  
  @ray.remote
  def child():
      ray.get(grandchild.remote())
  
  @ray.remote
  def finished_child():
      ray.put(1)
      return
  
  @ray.remote
  def parent():
      children = [child.remote() for _ in range(NUM_CHILD)]
      finished_children = ray.get([finished_child.remote() for _ in range(NUM_CHILD)])
      ray.get(children)
  
  ray.get(parent.remote())
  
  """
      proc = run_string_as_driver_nonblocking(script)
  
      def all_tasks_running():
          tasks = list_tasks()
          assert len(tasks) == 7, (
              "Incorrect number of tasks are reported. "
              "Expected length: 1 parent + 2 finished child +  2 failed child + "
              "2 failed grandchild tasks"
          )
          return True
  
      wait_for_condition(
          all_tasks_running,
          timeout=10,
          retry_interval_ms=500,
      )
      time_sleep_s = 3
      # Sleep for a while to allow driver job runs async.
      time.sleep(time_sleep_s)
  
      proc.kill()
  
      def verify():
          tasks = list_tasks(detail=True)
          assert len(tasks) == 7, (
              "Incorrect number of tasks are reported. "
              "Expected length: 1 parent + 2 finished child +  2 failed child + "
              "2 failed grandchild tasks"
          )
          for task in tasks:
              if "finished" in task["func_or_class_name"]:
                  assert (
                      task["state"] == "FINISHED"
                  ), f"task {task['func_or_class_name']} has wrong state"
              else:
                  assert (
                      task["state"] == "FAILED"
                  ), f"task {task['func_or_class_name']} has wrong state"
  
                  assert task["error_type"] == "WORKER_DIED"
                  assert "Job finishes" in task["error_message"]
  
                  duration_ms = (
                      task["end_time_ms"] - task["start_time_ms"]
                  )
                  assert (
                      # It takes time for the job to run
                      duration_ms > time_sleep_s / 2 * 1000
                      and duration_ms < 2 * time_sleep_s * 1000
                  )
  
          return True
  
      wait_for_condition(
          verify,
          timeout=10,
          retry_interval_ms=500,
      )
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 7 for calling a global function with example usage: test_fault_tolerance_job_failed(shutdown_only) and returns: True, True