code: def test_actor_fail(
      ray_start_4_cpus, cleanup, resource_manager_cls, where
  ):
      """Test that actor failures are handled properly.
  
      - Start actor that either fails on init or in a task (RayActorError)
      - Schedule task on actor
      - Assert that the correct callbacks are called
      """
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      # keep track of failed tasks and actors
      stats = Counter()
  
      @ray.remote
      class FailingActor:
          def __init__(self, where):
              self._where = where
              if self._where == "init":
                  raise RuntimeError("INIT")
  
          def fn(self):
              if self._where == "fn":
                  # SystemExit will invoke a RayActorError
                  raise SystemExit
              return True
  
      def fail_callback_actor(tracked_actor, exception):
          stats["failed_actor"] += 1
  
      def fail_callback_task(tracked_actor, exception):
          stats["failed_task"] += 1
  
      # Start actor
      tracked_actor = actor_manager.add_actor(
          cls=FailingActor,
          kwargs={"where": where},
          resource_request=ResourceRequest([{"CPU": 1}]),
          on_error=fail_callback_actor,
      )
  
      if where != "init":
          # Wait until it is started. This won't invoke any callback, yet
          actor_manager.next()
  
          assert stats["failed_actor"] == 0
          assert stats["failed_task"] == 0
  
          # Schedule task
          actor_manager.schedule_actor_task(
              tracked_actor, "fn", on_error=fail_callback_task
          )
  
      # Yield control and wait for task resolution. This will invoke the callback.
      actor_manager.next()
  
      assert stats["failed_actor"] == 1
      assert stats["failed_task"] == bool(where != "init")
  
imports: import threading
  import time
  from collections import Counter
  import gc
  from typing import Any, Optional, Type
  import pytest
  import ray
  from ray.air import ResourceRequest
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.air.execution._internal import Barrier
  from ray.air.execution._internal.actor_manager import RayActorManager
label_desc: Use image node_id 7 for calling a global function with example usage: test_actor_fail(ray_start_4_cpus, cleanup, resource_manager_cls, where) and returns: True