code: def test_worker_killer():
      ray.init()
  
      task_name = "worker_to_kill"
  
      @ray.remote
      def worker_to_kill():
          time.sleep(3)
  
      # Run WorkerKillerActor to kill 3 tasks, and run remote "worker_to_kill"
      # task with max_retries=3. 4 tasks in total (1 initial + 3 retries).
      # First 3 tasks will be killed, the last retry will succeed.
      worker_killer = get_and_run_resource_killer(
          WorkerKillerActor,
          1,
          max_to_kill=3,
          kill_filter_fn=lambda: lambda task: task.name == task_name,
      )
      worker_to_kill.options(name=task_name, max_retries=3).remote()
  
      def check():
          tasks = StateApiClient().list(
              StateResource.TASKS,
              options=ListApiOptions(
                  filters=[("name", "=", task_name)]
              ),
              raise_on_missing_output=False,
          )
          failed = 0
          finished = 0
          for task in tasks:
              if task.state == "FAILED":
                  failed += 1
              elif task.state == "FINISHED":
                  finished += 1
          return failed == 3 and finished == 1
  
      wait_for_condition(check, timeout=20)
  
      killed_tasks = ray.get(worker_killer.get_total_killed.remote())
      assert len(killed_tasks) == 3
  
      tasks = StateApiClient().list(
          StateResource.TASKS,
          options=ListApiOptions(filters=[("name", "=", task_name)]),
          raise_on_missing_output=False,
      )
      for task in tasks:
          if task.state == "FAILED":
              assert (task.task_id, task.worker_pid) in killed_tasks
  
      ray.shutdown()
  
imports: import sys
  import random
  import string
  import ray
  import pytest
  import time
  from ray.experimental import shuffle
  from ray.tests.conftest import _ray_start_chaos_cluster
  from ray.data._internal.progress_bar import ProgressBar
  from ray.util.placement_group import placement_group
  from ray._private.test_utils import NodeKillerActor, get_log_message, get_and_run_resource_killer, WorkerKillerActor, wait_for_condition
  from ray.exceptions import RayTaskError, ObjectLostError
  from ray.util.state.common import ListApiOptions, StateResource
  from ray.util.state.api import StateApiClient, list_nodes
  from ray.cluster_utils import AutoscalingCluster
label_desc: Use image node_id 8 for calling a global function with example usage: test_worker_killer() and returns: unknown