code: def test_exception_raised_when_actor_node_dies(
      ray_start_cluster_head,
  ):
      cluster = ray_start_cluster_head
      remote_node = cluster.add_node()
  
      @ray.remote(max_restarts=0, scheduling_strategy="SPREAD")
      class Counter:
          def __init__(self):
              self.x = 0
  
          def node_id(self):
              return ray._private.worker.global_worker.node.unique_id
  
          def inc(self):
              self.x += 1
              return self.x
  
      # Create an actor that is not on the raylet.
      actor = Counter.remote()
      while ray.get(actor.node_id.remote()) != remote_node.unique_id:
          actor = Counter.remote()
  
      # Kill the second node.
      cluster.remove_node(remote_node)
  
      # Submit some new actor tasks both before and after the node failure is
      # detected. Make sure that getting the result raises an exception.
      for _ in range(10):
          # Submit some new actor tasks.
          x_ids = [actor.inc.remote() for _ in range(5)]
          for x_id in x_ids:
              with pytest.raises(ray.exceptions.RayActorError):
                  # There is some small chance that ray.get will actually
                  # succeed (if the object is transferred before the raylet
                  # dies).
                  ray.get(x_id)
  
imports: import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  from ray.util.state import list_actors
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, convert_actor_state, kill_actor_and_wait_for_failure, make_global_state_accessor, run_string_as_driver, wait_for_condition, wait_for_pid_to_exit
  from ray._private.ray_constants import gcs_actor_scheduling_enabled
  from ray.experimental.internal_kv import _internal_kv_get, _internal_kv_put
label_desc: Use image node_id 6 for calling a global function with example usage: test_exception_raised_when_actor_node_dies(ray_start_cluster_head) and returns: ray, self