code: def test_kill_idle_workers_that_are_behind_owned_workers(
      shutdown_only,
  ):
      # When the first N idle workers own objects, and if we have N+N
      # total idle workers, we should make sure other N workers are killed.
      # It is because the idle workers are killed in the FIFO order.
      N = 4
      ray.init(
          num_cpus=1,
          _system_config={
              "kill_idle_workers_interval_ms": 10,
              "worker_lease_timeout_milliseconds": 0,
          },
      )
  
      @ray.remote
      def nested(i):
          if i >= (N * 2) - 1:
              return [ray.put(np.ones(1 * 1024 * 1024, dtype=np.uint8))]
          elif i >= N:
              return [
                  ray.put(np.ones(1 * 1024 * 1024, dtype=np.uint8))
              ] + ray.get(nested.remote(i + 1))
          else:
              return [1] + ray.get(nested.remote(i + 1))
  
      # The first N workers don't own objects
      # and the later N workers do.
      ref = ray.get(nested.remote(0))
      assert len(ref) == N * 2
      num_workers = len(get_workers())
      assert num_workers == N * 2
  
      # Make sure there are only N workers left after worker capping.
      wait_for_condition(lambda: len(get_workers()) == N)
  
file_path: ray/python/ray/tests/test_multi_tenancy.py
imports: import os
  import sys
  import time
  import pytest
  import numpy
  import ray
  from ray.core.generated import common_pb2
  from ray.core.generated import node_manager_pb2, node_manager_pb2_grpc
  from ray._private.test_utils import wait_for_condition, run_string_as_driver, run_string_as_driver_nonblocking
  from ray._private.utils import init_grpc_channel
label_desc: Use image node_id 10 for calling a global function with example usage: test_kill_idle_workers_that_are_behind_owned_workers(shutdown_only) and returns: list, unknown, unknown