code: def test_failed_task_error(shutdown_only):
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      # Test failed task with TASK_EXECUTION_EXCEPTION
      error_msg_str = "fail is expected to fail"
  
      @ray.remote
      def fail(x=None):
          if x is not None:
              time.sleep(x)
          raise ValueError(error_msg_str)
  
      with pytest.raises(ray.exceptions.RayTaskError):
          ray.get(fail.options(name="fail").remote())
  
      wait_for_condition(
          verify_failed_task,
          name="fail",
          error_type="TASK_EXECUTION_EXCEPTION",
          error_message=error_msg_str,
      )
  
      # Test canceled tasks with TASK_CANCELLED
      @ray.remote
      def not_running():
          raise ValueError("should not be run")
  
      with pytest.raises(ray.exceptions.TaskCancelledError):
          t = not_running.options(name="cancel-before-running").remote()
          ray.cancel(t)
          ray.get(t)
  
      # Cancel task doesn't have additional error message
      wait_for_condition(
          verify_failed_task,
          name="cancel-before-running",
          error_type="TASK_CANCELLED",
          error_message="",
      )
  
      # Test task failed when worker killed :WORKER_DIED
      @ray.remote(max_retries=0)
      def die():
          exit(27)
  
      with pytest.raises(ray.exceptions.WorkerCrashedError):
          ray.get(die.options(name="die-worker").remote())
  
      wait_for_condition(
          verify_failed_task,
          name="die-worker",
          error_type="WORKER_DIED",
          error_message="Worker exits with an exit code 27",
      )
  
      # Test actor task failed with actor dead: ACTOR_DIED
      @ray.remote
      class Actor:
          def f(self):
              time.sleep(999)
  
          def ready(self):
              pass
  
      a = Actor.remote()
      ray.get(a.ready.remote())
  
      with pytest.raises(ray.exceptions.RayActorError):
          ray.kill(a)
          ray.get(a.f.options(name="actor-killed").remote())
  
      wait_for_condition(
          verify_failed_task,
          name="actor-killed",
          error_type="ACTOR_DIED",
          error_message="The actor is dead because it was killed by `ray.kill`",
      )
  
imports: from collections import defaultdict
  from typing import Dict
  import os
  import pytest
  import sys
  import threading
  import time
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env import RuntimeEnv
  import ray
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_tasks
  from ray._private.worker import RayContext
label_desc: Use image node_id 3 for calling a global function with example usage: test_failed_task_error(shutdown_only) without return types