code: def test_replica_deletion_after_controller_recover(serve_instance):
      """Test that replicas are deleted when controller is recovered"""
  
      controller = serve.context._global_client._controller
  
      @serve.deployment(graceful_shutdown_timeout_s=3)
      class V1:
          async def __call__(self):
              while True:
                  await asyncio.sleep(0.1)
  
      handle = serve.run(V1.bind(), name="app")
      _ = handle.remote()
      serve.delete("app", _blocking=False)
  
      def check_replica(replica_state=None):
          id = DeploymentID("V1", "app")
          try:
              replicas = ray.get(
                  controller._dump_replica_states_for_testing.remote(id)
              )
          except RayTaskError as ex:
              # Deployment is not existed any more.
              if isinstance(ex, KeyError):
                  return []
              # Unexpected exception raised.
              raise ex
          if replica_state is None:
              replica_state = list(ReplicaState)
          else:
              replica_state = [replica_state]
          return replicas.get(replica_state)
  
      # Make sure the replica is in STOPPING state.
      wait_for_condition(
          lambda: len(check_replica(ReplicaState.STOPPING)) > 0
      )
      ray.kill(
          serve.context._global_client._controller, no_restart=False
      )
      # Make sure the replica is in STOPPING state.
      wait_for_condition(
          lambda: len(check_replica(ReplicaState.STOPPING)) > 0
      )
  
      # The graceful shutdown timeout of 3 seconds should be used
      wait_for_condition(lambda: len(check_replica()) == 0, timeout=20)
      # Application should be removed soon after
      wait_for_condition(
          lambda: "app" not in serve.status().applications, timeout=20
      )
  
imports: import asyncio
  import logging
  import os
  import re
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.exceptions import RayTaskError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_CONTROLLER_NAME, SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE, SERVE_PROXY_NAME
  from ray.serve._private.utils import get_random_string
  from ray.serve.schema import LoggingConfig
  from ray.serve.tests.test_failure import request_with_retries
  from ray.util.state import list_actors
label_desc: Use image node_id 4 for calling a global function with example usage: test_replica_deletion_after_controller_recover(serve_instance) and returns: replicas, list