code: def test_task_failure_when_driver_local_raylet_dies(
      ray_start_cluster,
  ):
      cluster = ray_start_cluster
      head = cluster.add_node(num_cpus=4, resources={"foo": 1})
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      @ray.remote(resources={"foo": 1})
      def func():
          internal_kv._internal_kv_put("test_func", "func")
          while True:
              time.sleep(1)
  
      func.remote()
      while not internal_kv._internal_kv_exists("test_func"):
          time.sleep(0.1)
  
      # The lease request should wait inside raylet
      # since there is no available resources.
      ret = func.options(name="task-local-raylet-dead").remote()
  
      # Waiting for the lease request to reach raylet.
      def task_running():
          tasks = list_tasks(
              filters=[("name", "=", "task-local-raylet-dead")]
          )
          assert len(tasks) == 1
          assert tasks[0]["state"] == "PENDING_NODE_ASSIGNMENT"
          return True
  
      wait_for_condition(task_running)
  
      head.kill_raylet()
      with pytest.raises(LocalRayletDiedError):
          ray.get(ret)
  
      # Check the task failure states for observability.
      wait_for_condition(
          verify_failed_task,
          name="task-local-raylet-dead",
          error_type="LOCAL_RAYLET_DIED",
          error_message="The worker failed to receive a response from the local raylet",
      )
  
imports: import subprocess
  import sys
  import time
  import grpc
  import numpy
  import psutil
  import pytest
  from grpc._channel import _InactiveRpcError
  from ray.util.state import list_tasks
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  import ray._private.ray_constants
  import ray.experimental.internal_kv
  from ray import NodeID
  from ray._private.test_utils import SignalActor, get_error_message, init_error_pubsub, run_string_as_driver, wait_for_condition, kill_raylet
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.core.generated import gcs_service_pb2, gcs_service_pb2_grpc, node_manager_pb2, node_manager_pb2_grpc
  from ray.exceptions import LocalRayletDiedError
label_desc: Use image node_id 11 for calling a global function with example usage: test_task_failure_when_driver_local_raylet_dies(ray_start_cluster) and returns: True