code: def test_leaked_pg_removed_on_controller_recovery(serve_instance):
      """Verify that leaked PGs are removed on controller recovery.
  
      A placement group can be "leaked" if the replica is killed while the controller is
      down or the controller crashes between creating a placement group and its replica.
  
      In these cases, the controller should detect the leak on recovery and delete the
      leaked placement group(s).
      """
  
      @serve.deployment(
          placement_group_bundles=[{"CPU": 1}],
          health_check_period_s=0.1,
      )
      class D:
          def die(self):
              os._exit(1)
  
          def get_pg(self) -> PlacementGroup:
              return get_current_placement_group()
  
      h = serve.run(D.bind(), name="pg_test")
  
      prev_pg = h.get_pg.remote().result()
      assert len(get_all_live_placement_group_names()) == 1
  
      # Kill the controller and the replica immediately after.
      # This will cause the controller to *not* detect the replica on recovery, but it
      # should still detect the leaked placement group and clean it up.
      ray.kill(_get_global_client()._controller, no_restart=False)
      with pytest.raises(ray.exceptions.RayActorError):
          h.die.remote().result()
  
      def leaked_pg_cleaned_up():
          try:
              new_pg = h.get_pg.remote().result()
          except ray.exceptions.RayActorError:
              return False
  
          return (
              len(get_all_live_placement_group_names()) == 1
              and new_pg != prev_pg
          )
  
      # Verify that a new replica is placed with a new placement group and the old
      # placement group has been removed.
      wait_for_condition(leaked_pg_cleaned_up)
  
      serve.delete("pg_test")
      assert len(get_all_live_placement_group_names()) == 0
  
file_path: ray/python/ray/serve/tests/test_replica_placement_group.py
imports: import os
  import sys
  import pytest
  import ray
  from ray import serve
  from ray._private.test_utils import wait_for_condition
  from ray.serve._private.utils import get_all_live_placement_group_names
  from ray.serve.context import _get_global_client
  from ray.util.placement_group import PlacementGroup, get_current_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 7 for calling a global function with example usage: test_leaked_pg_removed_on_controller_recovery(serve_instance) and returns: unknown, get_current_placement_group, False