code: def test_kv_timeout(ray_start_regular):
      gcs_address = ray._private.worker.global_worker.gcs_client.address
      gcs_client = ray._raylet.GcsClient(
          address=gcs_address, nums_reconnect_retry=0
      )
  
      assert gcs_client.internal_kv_put(b"A", b"", False, b"") == 1
  
      with stop_gcs_server():
          with pytest.raises(
              ray.exceptions.RpcError, match="Deadline Exceeded"
          ):
              gcs_client.internal_kv_put(
                  b"A", b"B", False, b"NS", timeout=2
              )
  
          with pytest.raises(
              ray.exceptions.RpcError, match="Deadline Exceeded"
          ):
              gcs_client.internal_kv_get(b"A", b"NS", timeout=2)
  
          with pytest.raises(
              ray.exceptions.RpcError, match="Deadline Exceeded"
          ):
              gcs_client.internal_kv_keys(b"A", b"NS", timeout=2)
  
          with pytest.raises(
              ray.exceptions.RpcError, match="Deadline Exceeded"
          ):
              gcs_client.internal_kv_del(b"A", True, b"NS", timeout=2)
  
imports: import contextlib
  import os
  import time
  import signal
  import sys
  import asyncio
  import async_timeout
  import pytest
  import ray
  import redis
  from ray._raylet import GcsClient
  import ray._private.gcs_utils
  from ray._private.test_utils import enable_external_redis, find_free_port, generate_system_config_map, async_wait_for_condition_async_predicate
  import ray._private.ray_constants
label_desc: Use image node_id 3 for calling a global function with example usage: test_kv_timeout(ray_start_regular) without return types