code: def test_error_custom(ray_start_4_cpus, resource_manager_cls):
      """When an `on_error` callback is specified, it is invoked."""
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      stats = Counter()
  
      def error_callback(tracked_actor, exception):
          stats["exception"] += 1
  
      tracked_actor = actor_manager.add_actor(
          cls=Actor,
          kwargs={},
          resource_request=ResourceRequest([{"CPU": 4}]),
      )
      actor_manager.schedule_actor_task(
          tracked_actor, "foo", (1, True), on_error=error_callback
      )
  
      actor_manager.next()
      actor_manager.next()
      assert stats["exception"] == 1
  
file_path: ray/python/ray/air/tests/execution/test_tracked_actor_task.py
imports: from collections import Counter
  import pytest
  import ray
  from ray.air import ResourceRequest
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  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_error_custom(ray_start_4_cpus, resource_manager_cls) without return types