code: def test_remove_placement_group_worker_startup_slowly(
      set_runtime_env_plugins, ray_start_cluster
  ):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      ray.init(address=cluster.address)
  
      placement_group = ray.util.placement_group(
          [{"CPU": 2}, {"CPU": 2}]
      )
      assert placement_group.wait(10)
  
      @ray.remote(num_cpus=2)
      class A:
          def f(self):
              return 3
  
      @ray.remote(num_cpus=2, max_retries=0)
      def long_running_task():
          print(os.getpid())
          import time
  
          time.sleep(60)
  
      # Schedule a long-running task that uses
      # runtime env to mock worker start up slowly.
      task_ref = long_running_task.options(
          scheduling_strategy=PlacementGroupSchedulingStrategy(
              placement_group=placement_group
          ),
          runtime_env={MOCK_WORKER_STARTUP_SLOWLY_PLUGIN_NAME: {}},
      ).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)
  
      # Make sure the actor has been killed
      # because of the removal of the pg.
      # TODO(@clay4444): Make it throw a `ActorPlacementGroupRemoved`.
      with pytest.raises(
          ray.exceptions.RayActorError, match="actor died"
      ):
          ray.get(a.f.remote(), timeout=3.0)
  
      # The long-running task should still be in the state
      # of leasing-worker bacause of the worker startup delay.
      with pytest.raises(ray.exceptions.TaskPlacementGroupRemoved):
          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 2 for calling a global function with example usage: test_remove_placement_group_worker_startup_slowly(set_runtime_env_plugins, ray_start_cluster) and returns: int