code: def test_raylet_graceful_shutdown_through_rpc(
      ray_start_cluster_head, error_pubsub
  ):
      """
      Prepare the cluster.
      """
      cluster = ray_start_cluster_head
      head_node_port = None
      for n in ray.nodes():
          head_node_port = int(n["NodeManagerPort"])
      worker = cluster.add_node(num_cpus=1)
      cluster.wait_for_nodes()
      worker_node_port = None
      for n in ray.nodes():
          port = int(n["NodeManagerPort"])
          if port != head_node_port and n["alive"]:
              worker_node_port = port
      """
      warm up the cluster
      """
  
      @ray.remote
      def f():
          pass
  
      ray.get(f.remote())
  
      # Kill a raylet gracefully.
      def kill_raylet(ip, port, graceful=True):
          raylet_address = f"{ip}:{port}"
          channel = grpc.insecure_channel(raylet_address)
          stub = node_manager_pb2_grpc.NodeManagerServiceStub(channel)
          print(f"Sending a shutdown request to {ip}:{port}")
          try:
              stub.ShutdownRaylet(
                  node_manager_pb2.ShutdownRayletRequest(
                      graceful=graceful
                  )
              )
          except _InactiveRpcError:
              assert not graceful
  
      """
      Kill the first worker ungracefully.
      """
      ip = worker.node_ip_address
      kill_raylet(ip, worker_node_port, graceful=False)
      p = error_pubsub
      errors = get_error_message(
          p, 1, ray_constants.REMOVED_NODE_ERROR, timeout=10
      )
      # Should print the heartbeat messages.
      assert (
          "has missed too many heartbeats from it"
          in errors[0]["error_message"]
      )
      # NOTE the killed raylet is a zombie since the
      # parent process (the pytest script) hasn't called wait syscall.
      # For normal scenarios where raylet is created by
      # ray start, this issue won't exist.
      try:
          wait_for_condition(lambda: len(search_raylet(cluster)) == 1)
      except Exception:
          print("More than one raylets are detected.")
          print(search_raylet(cluster))
      """
      Kill the second worker gracefully.
      """
      worker = cluster.add_node(num_cpus=0)
      worker_node_port = None
      for n in ray.nodes():
          port = int(n["NodeManagerPort"])
          if port != head_node_port and n["alive"]:
              worker_node_port = port
      # Kill the second worker gracefully.
      ip = worker.node_ip_address
      kill_raylet(ip, worker_node_port, graceful=True)
      p = error_pubsub
      # Error shouldn't be printed to the driver.
      errors = get_error_message(
          p, 1, ray_constants.REMOVED_NODE_ERROR, timeout=5
      )
      # Error messages shouldn't be published.
      assert len(errors) == 0
      try:
          wait_for_condition(lambda: len(search_raylet(cluster)) == 1)
      except Exception:
          print("More than one raylets are detected.")
          print(search_raylet(cluster))
      """
      Make sure head node is not dead.
      """
      ray.get(f.options(num_cpus=0).remote())
  
imports: import subprocess
  import sys
  import time
  import grpc
  import numpy
  import psutil
  import pytest
  from grpc._channel import _InactiveRpcError
  from ray.util.state import list_tasks
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  import ray._private.ray_constants
  import ray.experimental.internal_kv
  from ray import NodeID
  from ray._private.test_utils import SignalActor, get_error_message, init_error_pubsub, run_string_as_driver, wait_for_condition, kill_raylet
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.core.generated import gcs_service_pb2, gcs_service_pb2_grpc, node_manager_pb2, node_manager_pb2_grpc
  from ray.exceptions import LocalRayletDiedError
label_desc: Use image node_id 8 for calling a global function with example usage: test_raylet_graceful_shutdown_through_rpc(ray_start_cluster_head, error_pubsub) without return types