code: def test_automatic_cleanup_job(ray_start_cluster):
      # Make sure the placement groups created by a
      # job, actor, and task are cleaned when the job is done.
      cluster = ray_start_cluster
      num_nodes = 3
      num_cpu_per_node = 4
      # Create 3 nodes cluster.
      for _ in range(num_nodes):
          cluster.add_node(num_cpus=num_cpu_per_node)
      cluster.wait_for_nodes()
  
      info = ray.init(address=cluster.address)
      available_cpus = ray.available_resources()["CPU"]
      assert available_cpus == num_nodes * num_cpu_per_node
  
      driver_code = f"""
  import ray
  
  ray.init(address="{info["address"]}")
  
  def create_pg():
      pg = ray.util.placement_group(
              [{{"CPU": 1}} for _ in range(3)],
              strategy="STRICT_SPREAD")
      ray.get(pg.ready())
      return pg
  
  @ray.remote(num_cpus=0)
  def f():
      create_pg()
  
  @ray.remote(num_cpus=0)
  class A:
      def create_pg(self):
          create_pg()
  
  ray.get(f.remote())
  a = A.remote()
  ray.get(a.create_pg.remote())
  # Create 2 pgs to make sure multiple placement groups that belong
  # to a single job will be properly cleaned.
  create_pg()
  create_pg()
  
  ray.shutdown()
      """
  
      run_string_as_driver(driver_code)
  
      # Wait until the driver is reported as dead by GCS.
      def is_job_done():
          jobs = ray._private.state.jobs()
          for job in jobs:
              if job["IsDead"]:
                  return True
          return False
  
      def assert_num_cpus(expected_num_cpus):
          if expected_num_cpus == 0:
              return "CPU" not in ray.available_resources()
          return ray.available_resources()["CPU"] == expected_num_cpus
  
      wait_for_condition(is_job_done)
      available_cpus = ray.available_resources()["CPU"]
      wait_for_condition(
          lambda: assert_num_cpus(num_nodes * num_cpu_per_node)
      )
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, get_error_message, get_other_nodes, kill_actor_and_wait_for_failure, placement_group_assert_no_leak, run_string_as_driver, wait_for_condition
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 10 for calling a global function with example usage: test_automatic_cleanup_job(ray_start_cluster) and returns: False, unknown, unknown, True