code: def test_detached_actor_cleanup_due_to_failure(ray_start_cluster):
      cluster = ray_start_cluster
      node = cluster.add_node(resources={"second_node": 1})
      cluster.wait_for_nodes()
  
      @ray.remote
      class DetachedActor:
          def ping(self):
              return "pong"
  
          def kill_itself(self):
              # kill itself.
              os._exit(0)
  
      worker_failure_actor_name = "worker_failure_actor_name"
      node_failure_actor_name = "node_failure_actor_name"
  
      def wait_until_actor_dead(handle):
          actor_status = ray._private.state.actors(
              actor_id=handle._actor_id.hex()
          )
          max_wait_time = 10
          wait_time = 0
          while actor_status["State"] != convert_actor_state(
              gcs_utils.ActorTableData.DEAD
          ):
              actor_status = ray._private.state.actors(
                  actor_id=handle._actor_id.hex()
              )
              time.sleep(1.0)
              wait_time += 1
              if wait_time >= max_wait_time:
                  assert (
                      None
                  ), "It took too much time to kill an actor: {}".format(
                      handle._actor_id
                  )
  
      def create_detached_actor_blocking(
          actor_name, schedule_in_second_node=False
      ):
          resources = (
              {"second_node": 1}
              if schedule_in_second_node
              else {"first_node": 1}
          )
          actor_handle = DetachedActor.options(
              lifetime="detached", name=actor_name, resources=resources
          ).remote()
          # Wait for detached actor creation.
          assert ray.get(actor_handle.ping.remote()) == "pong"
          return actor_handle
  
      # Name should be cleaned when workers fail
      deatched_actor = create_detached_actor_blocking(
          worker_failure_actor_name
      )
      deatched_actor.kill_itself.remote()
      wait_until_actor_dead(deatched_actor)
      # Name should be available now.
      deatched_actor = create_detached_actor_blocking(
          worker_failure_actor_name
      )
      assert ray.get(deatched_actor.ping.remote()) == "pong"
  
      # Name should be cleaned when nodes fail.
      deatched_actor = create_detached_actor_blocking(
          node_failure_actor_name, schedule_in_second_node=True
      )
      cluster.remove_node(node)
      wait_until_actor_dead(deatched_actor)
      # Name should be available now.
      deatched_actor = create_detached_actor_blocking(
          node_failure_actor_name
      )
      assert ray.get(deatched_actor.ping.remote()) == "pong"
  
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 21 for calling a global function with example usage: test_detached_actor_cleanup_due_to_failure(ray_start_cluster) and returns: actor_handle, str