code: def test_start_stop_actor(
      ray_start_4_cpus, resource_manager_cls, actor_cls, kill
  ):
      """Test that starting and stopping actors work and invokes a callback.
  
      - Start an actor
      - Starting should trigger start callback
      - Schedule actor task, which should resolve (meaning actor successfully started)
      - Stop actor, which should resolve and trigger stop callback
      - Schedule remote fn that takes up all cluster resources. This should resolve,
        meaning that the actor was stopped successfully.
      """
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      # Start actor, set callbacks
      tracked_actor = actor_manager.add_actor(
          cls=actor_cls,
          kwargs={"key": "val"},
          resource_request=ResourceRequest([{"CPU": 4}]),
          on_start=_raise(Started),
          on_stop=_raise(Stopped),
          on_error=_raise(Failed),
      )
  
      # Actor should be started
      with pytest.raises(Started):
          actor_manager.next()
  
      # Schedule task on actor which should resolve (actor successfully started)
      actor_manager.schedule_actor_task(
          tracked_actor, "task", (1,), on_result=_raise(Result)
      )
  
      with pytest.raises(Result):
          actor_manager.next()
  
      # Now we can assert that there are no CPUS resources available anymore.
      # Note that actor starting is asynchronous, so we can't assert this right away
      # - that's why we wait for the actor task to resolve first.
      assert (
          ray.available_resources().get("CPU", 0.0) == 0
      ), ray.available_resources()
  
      # Stop actor
      actor_manager.remove_actor(tracked_actor, kill=kill)
  
      with pytest.raises(Stopped):
          actor_manager.next()
  
      # This task takes up all the cluster resources. It should resolve now that
      # the actor was terminated.
      assert ray.get(fn.remote(), timeout=5)
  
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 5 for calling a global function with example usage: test_start_stop_actor(ray_start_4_cpus, resource_manager_cls, actor_cls, kill) without return types