code: def test_placement_group_reschedule_when_node_dead(
      ray_start_cluster, connect_to_client
  ):
      @ray.remote(num_cpus=1)
      class Actor(object):
          def __init__(self):
              self.n = 0
  
          def value(self):
              return self.n
  
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
      cluster.add_node(num_cpus=4)
      cluster.wait_for_nodes()
      ray.init(
          address=cluster.address, namespace="default_test_namespace"
      )
  
      # Make sure both head and worker node are alive.
      nodes = ray.nodes()
      assert len(nodes) == 3
      assert (
          nodes[0]["alive"] and nodes[1]["alive"] and nodes[2]["alive"]
      )
  
      with connect_to_client_or_not(connect_to_client):
          placement_group = ray.util.placement_group(
              name="name",
              strategy="SPREAD",
              bundles=[{"CPU": 2}, {"CPU": 2}, {"CPU": 2}],
          )
          actor_1 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=0,
              ),
              lifetime="detached",
          ).remote()
          actor_2 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=1,
              ),
              lifetime="detached",
          ).remote()
          actor_3 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=2,
              ),
              lifetime="detached",
          ).remote()
          ray.get(actor_1.value.remote())
          ray.get(actor_2.value.remote())
          ray.get(actor_3.value.remote())
  
          cluster.remove_node(
              get_other_nodes(cluster, exclude_head=True)[-1]
          )
          cluster.wait_for_nodes()
  
          actor_4 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=0,
              ),
              lifetime="detached",
          ).remote()
          actor_5 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=1,
              ),
              lifetime="detached",
          ).remote()
          actor_6 = Actor.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=placement_group,
                  placement_group_bundle_index=2,
              ),
              lifetime="detached",
          ).remote()
          ray.get(actor_4.value.remote())
          ray.get(actor_5.value.remote())
          ray.get(actor_6.value.remote())
          placement_group_assert_no_leak([placement_group])
          ray.shutdown()
  
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 7 for calling a global function with example usage: test_placement_group_reschedule_when_node_dead(ray_start_cluster, connect_to_client) and returns: self