code: def test_failed_task(ray_start_shared_local_modes, error_pubsub):
      @ray.remote
      def throw_exception_fct1():
          raise Exception("Test function 1 intentionally failed.")
  
      @ray.remote
      def throw_exception_fct2():
          raise Exception("Test function 2 intentionally failed.")
  
      @ray.remote(num_returns=3)
      def throw_exception_fct3(x):
          raise Exception("Test function 3 intentionally failed.")
  
      p = error_pubsub
  
      throw_exception_fct1.remote()
      throw_exception_fct1.remote()
  
      if (
          ray._private.worker.global_worker.mode
          != ray._private.worker.LOCAL_MODE
      ):
          msgs = get_error_message(
              p, 2, ray._private.ray_constants.TASK_PUSH_ERROR
          )
          assert len(msgs) == 2
          for msg in msgs:
              assert (
                  "Test function 1 intentionally failed."
                  in msg["error_message"]
              )
  
      x = throw_exception_fct2.remote()
      try:
          ray.get(x)
      except Exception as e:
          assert "Test function 2 intentionally failed." in str(e)
      else:
          # ray.get should throw an exception.
          assert False
  
      x, y, z = throw_exception_fct3.remote(1.0)
      for ref in [x, y, z]:
          try:
              ray.get(ref)
          except Exception as e:
              assert "Test function 3 intentionally failed." in str(e)
          else:
              # ray.get should throw an exception.
              assert False
  
      class CustomException(ValueError):
          def __init__(self, msg):
              super().__init__(msg)
              self.field = 1
  
          def f(self):
              return 2
  
      @ray.remote
      def f():
          raise CustomException("This function failed.")
  
      try:
          ray.get(f.remote())
      except Exception as e:
          assert "This function failed." in str(e)
          assert isinstance(e, ValueError)
          assert isinstance(e, CustomException)
          assert isinstance(e, ray.exceptions.RayTaskError)
          assert "RayTaskError(CustomException)" in repr(e)
          assert e.field == 1
          assert e.f() == 2
      else:
          # ray.get should throw an exception.
          assert False
  
imports: import logging
  import os
  import pickle
  import random
  import sys
  import time
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import SignalActor, client_test_enabled, get_error_message, run_string_as_driver
label_desc: Use image node_id 31 for calling a global function with example usage: test_failed_task(ray_start_shared_local_modes, error_pubsub) and returns: int