code: def test_connect_with_disconnected_node(shutdown_only):
      config = {
          "health_check_failure_threshold": 50,
          "health_check_period_ms": 10,
          "health_check_initial_delay_ms": 0,
      }
      cluster = Cluster()
      cluster.add_node(num_cpus=0, _system_config=config)
      ray.init(address=cluster.address)
      p = init_error_pubsub()
      errors = get_error_message(p, 1, timeout=5)
      assert len(errors) == 0
      # This node will be killed by SIGKILL, ray_monitor will mark it to dead.
      dead_node = cluster.add_node(num_cpus=0)
      cluster.remove_node(dead_node, allow_graceful=False)
      errors = get_error_message(p, 1, ray_constants.REMOVED_NODE_ERROR)
      assert len(errors) == 1
      # This node is killed by SIGKILL, ray_monitor will mark it to dead.
      dead_node = cluster.add_node(num_cpus=0)
      cluster.remove_node(dead_node, allow_graceful=False)
      errors = get_error_message(p, 1, ray_constants.REMOVED_NODE_ERROR)
      assert len(errors) == 1
      # This node is killed by SIGTERM, ray_monitor will not mark it again.
      removing_node = cluster.add_node(num_cpus=0)
      cluster.remove_node(removing_node, allow_graceful=True)
      errors = get_error_message(
          p, 1, ray_constants.REMOVED_NODE_ERROR, timeout=2
      )
      assert len(errors) == 0
      # There is no connection error to a dead node.
      errors = get_error_message(p, 1, timeout=2)
      assert len(errors) == 0
      p.close()
  
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 5 for calling a global function with example usage: test_connect_with_disconnected_node(shutdown_only) without return types