code: def test_error_isolation(call_ray_start):
      address = call_ray_start
      # Connect a driver to the Ray cluster.
      ray.init(address=address)
  
      # If a GRPC call exceeds timeout, the calls is cancelled at client side but
      # server may still reply to it, leading to missed message. Using a sequence
      # number to ensure no message is dropped can be the long term solution,
      # but its complexity and the fact the Ray subscribers do not use deadline
      # in production makes it less preferred.
      # Therefore, a simpler workaround is used instead: a different subscriber
      # is used for each get_error_message() call.
      subscribers = [init_error_pubsub() for _ in range(3)]
  
      # There shouldn't be any errors yet.
      errors = get_error_message(subscribers[0], 1, timeout=2)
      assert len(errors) == 0
  
      error_string1 = "error_string1"
      error_string2 = "error_string2"
  
      @ray.remote
      def f():
          raise Exception(error_string1)
  
      # Run a remote function that throws an error.
      with pytest.raises(Exception):
          ray.get(f.remote())
  
      # Wait for the error to appear in Redis.
      errors = get_error_message(subscribers[1], 1)
  
      # Make sure we got the error.
      assert len(errors) == 1
      assert error_string1 in errors[0]["error_message"]
  
      # Start another driver and make sure that it does not receive this
      # error. Make the other driver throw an error, and make sure it
      # receives that error.
      driver_script = """
  import ray
  import time
  from ray._private.test_utils import init_error_pubsub, get_error_message
  
  ray.init(address="{}")
  subscribers = [init_error_pubsub() for _ in range(2)]
  time.sleep(1)
  errors = get_error_message(subscribers[0], 1, timeout=2)
  assert len(errors) == 0
  
  @ray.remote
  def f():
      raise Exception("{}")
  
  try:
      ray.get(f.remote())
  except Exception as e:
      pass
  
  errors = get_error_message(subscribers[1], 1)
  assert len(errors) == 1
  
  assert "{}" in errors[0]["error_message"]
  
  print("success")
  """.format(
          address, error_string2, error_string2
      )
  
      out = run_string_as_driver(driver_script)
      # Make sure the other driver succeeded.
      assert "success" in out
  
      # Make sure that the other error message doesn't show up for this
      # driver.
      errors = get_error_message(subscribers[2], 1)
      assert len(errors) == 1
  
file_path: ray/python/ray/tests/test_multi_node.py
imports: import os
  import sys
  import time
  import psutil
  import pytest
  import ray
  from ray._private import ray_constants
  from ray._private.test_utils import RayTestTimeoutException, get_error_message, init_error_pubsub, object_memory_usage, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
label_desc: Use image node_id 2 for calling a global function with example usage: test_error_isolation(call_ray_start) without return types