code: def test_remove_placement_group(ray_start_cluster, connect_to_client):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      ray.init(address=cluster.address)
  
      @ray.remote
      def warmup():
          pass
  
      # warm up the cluster.
      ray.get([warmup.remote() for _ in range(4)])
  
      with connect_to_client_or_not(connect_to_client):
          # First try to remove a placement group that doesn't
          # exist. This should not do anything.
          random_group_id = PlacementGroupID.from_random()
          random_placement_group = PlacementGroup(random_group_id)
          for _ in range(3):
              ray.util.remove_placement_group(random_placement_group)
  
          # Creating a placement group as soon as it is
          # created should work.
          placement_group = ray.util.placement_group(
              [{"CPU": 2}, {"CPU": 2}]
          )
          assert placement_group.wait(10)
  
          ray.util.remove_placement_group(placement_group)
  
          wait_for_condition(
              lambda: is_placement_group_removed(placement_group)
          )
  
          # # Now let's create a placement group.
          placement_group = ray.util.placement_group(
              [{"CPU": 2}, {"CPU": 2}]
          )
          assert placement_group.wait(10)
  
          # Create an actor that occupies resources.
          @ray.remote(num_cpus=2)
          class A:
              def f(self):
                  return 3
  
          # Currently, there's no way to prevent
          # tasks to be retried for removed placement group.
          # Set max_retrie=0 for testing.
          # TODO(sang): Handle this edge case.
          @ray.remote(num_cpus=2, max_retries=0)
          def long_running_task():
              print(os.getpid())
              import time
  
              time.sleep(50)
  
          # Schedule a long running task and actor.
          task_ref = long_running_task.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group
              )
          ).remote()
          a = A.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group
              )
          ).remote()
          assert ray.get(a.f.remote()) == 3
  
          ray.util.remove_placement_group(placement_group)
          # Subsequent remove request shouldn't do anything.
          for _ in range(3):
              ray.util.remove_placement_group(placement_group)
  
          # Make sure placement group resources are
          # released and we can schedule this task.
          @ray.remote(num_cpus=4)
          def f():
              return 3
  
          assert ray.get(f.remote()) == 3
          # Since the placement group is removed,
          # the actor should've been killed.
          # That means this request should fail.
          with pytest.raises(
              ray.exceptions.RayActorError, match="actor died"
          ):
              ray.get(a.f.remote(), timeout=3.0)
          with pytest.raises(ray.exceptions.WorkerCrashedError):
              ray.get(task_ref)
  
imports: import pytest
  import os
  import sys
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import get_other_nodes, wait_for_condition, is_placement_group_removed, placement_group_assert_no_leak
  from ray._raylet import PlacementGroupID
  from ray.util.placement_group import PlacementGroup
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray._private.runtime_env.context import RuntimeEnvContext
  from ray._private.runtime_env.plugin import RuntimeEnvPlugin
label_desc: Use image node_id 1 for calling a global function with example usage: test_remove_placement_group(ray_start_cluster, connect_to_client) and returns: int, int