code: def test_pg_removed_on_replica_crash(serve_instance):
      """Verify that PGs are removed when a replica crashes unexpectedly."""
  
      @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")
  
      # Get the placement group for the original replica.
      assert len(get_all_live_placement_group_names()) == 1
      pg = h.get_pg.remote().result()
  
      # Kill the replica forcefully.
      with pytest.raises(ray.exceptions.RayActorError):
          h.die.remote().result()
  
      def new_replica_scheduled():
          try:
              h.get_pg.remote().result()
          except ray.exceptions.RayActorError:
              return False
  
          return True
  
      # The original placement group should be deleted and a new replica should
      # be scheduled with its own new placement group.
      wait_for_condition(new_replica_scheduled)
      new_pg = h.get_pg.remote().result()
      assert pg != new_pg
      assert len(get_all_live_placement_group_names()) == 1
  
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 5 for calling a global function with example usage: test_pg_removed_on_replica_crash(serve_instance) and returns: True, get_current_placement_group, False