code: def test_failed_task_failed_due_to_node_failure(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
      node = cluster.add_node(num_cpus=2)
  
      driver_script = """
  import ray
  ray.init("auto")
  
  @ray.remote(num_cpus=2, max_retries=0)
  def sleep():
      import time
      time.sleep(999)
  
  x = sleep.options(name="node-killed").remote()
  ray.get(x)
      """
  
      run_string_as_driver_nonblocking(driver_script)
  
      def driver_running():
          t = list_tasks(filters=[("name", "=", "node-killed")])
          return len(t) > 0
  
      wait_for_condition(driver_running)
  
      # Kill the node
      cluster.remove_node(node)
  
      wait_for_condition(
          verify_failed_task,
          name="node-killed",
          error_type="NODE_DIED",
          error_message="Task failed due to the node dying",
      )
  
imports: from collections import defaultdict
  from typing import Dict
  import os
  import pytest
  import sys
  import threading
  import time
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env import RuntimeEnv
  import ray
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_tasks
  from ray._private.worker import RayContext
label_desc: Use image node_id 4 for calling a global function with example usage: test_failed_task_failed_due_to_node_failure(ray_start_cluster) and returns: unknown