code: def test_actor_failover_with_bad_network(ray_start_cluster_head):
      # The test case is to cover the scenario that when an actor FO happens,
      # the caller receives the actor ALIVE notification and connects to the new
      # actor instance while there are still some tasks sent to the previous
      # actor instance haven't returned.
      #
      # It's not easy to reproduce this scenario, so we set
      # `raylet_death_check_interval_milliseconds` to a large value and add a
      # never-return function for the actor to keep the RPC connection alive
      # while killing the node to trigger actor failover. Later we send SIGKILL
      # to kill the previous actor process to let the task fail.
      #
      # The expected behavior is that after the actor is alive again and the
      # previous RPC connection is broken, tasks sent via the previous RPC
      # connection should fail but tasks sent via the new RPC connection should
      # succeed.
  
      cluster = ray_start_cluster_head
      node = cluster.add_node(num_cpus=1)
  
      @ray.remote(max_restarts=1)
      class Actor:
          def getpid(self):
              return os.getpid()
  
          def never_return(self):
              while True:
                  time.sleep(1)
              return 0
  
      # The actor should be placed on the non-head node.
      actor = Actor.remote()
      pid = ray.get(actor.getpid.remote())
  
      # Submit a never-return task (task 1) to the actor. The return
      # object should be unready.
      obj1 = actor.never_return.remote()
      with pytest.raises(GetTimeoutError):
          ray.get(obj1, timeout=1)
  
      # Kill the non-head node and start a new one. Now GCS should trigger actor
      # FO. Since we changed the interval of worker checking death of Raylet,
      # the actor process won't quit in a short time.
      cluster.remove_node(node, allow_graceful=False)
      cluster.add_node(num_cpus=1)
  
      # The removed node will be marked as dead by GCS after 1 second and task 1
      # will return with failure after that.
      with pytest.raises(RayActorError):
          ray.get(obj1, timeout=2)
  
      # Wait for the actor to be alive again in a new worker process.
      def check_actor_restart():
          actors = list(ray._private.state.actors().values())
          assert len(actors) == 1
          print(actors)
          return (
              actors[0]["State"]
              == convert_actor_state(gcs_utils.ActorTableData.ALIVE)
              and actors[0]["NumRestarts"] == 1
          )
  
      wait_for_condition(check_actor_restart)
  
      # Kill the previous actor process.
      os.kill(pid, signal.SIGKILL)
  
      # Submit another task (task 2) to the actor.
      obj2 = actor.getpid.remote()
  
      # We should be able to get the return value of task 2 without any issue
      ray.get(obj2)
  
imports: import os
  import signal
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.test_utils import SignalActor, convert_actor_state, get_error_message, init_error_pubsub, wait_for_condition
  from ray.exceptions import GetTimeoutError, RayActorError, RayTaskError
label_desc: Use image node_id 20 for calling a global function with example usage: test_actor_failover_with_bad_network(ray_start_cluster_head) and returns: unknown, os, int