code: def test_worker_exit_system_error(ray_start_cluster):
      """
      SYSTEM_ERROR
      - (tested) Failure from the connection E.g., core worker dead.
      - (tested) Unexpected exception or exit with exit_code !=0 on core worker.
      - (tested for owner node death) Node died. Currently worker failure detection
          upon node death is not detected by Ray. TODO(sang): Fix it.
      - (Cannot test) Direct call failure.
      """
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=1, resources={"worker": 1})
  
      @ray.remote
      class Actor:
          def pid(self):
              import os
  
              return os.getpid()
  
          def exit(self, exit_code):
              sys.exit(exit_code)
  
      """
      Failure from the connection
      """
      a = Actor.remote()
      pid = ray.get(a.pid.remote())
      print(pid)
      os.kill(pid, signal.SIGKILL)
  
      def verify_connection_failure():
          worker = get_worker_by_pid(pid)
          print(worker)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          # If the worker is killed by SIGKILL, it is highly likely by OOM, so
          # the error message should contain information.
          return type == "SYSTEM_ERROR" and "OOM" in detail
  
      wait_for_condition(verify_connection_failure)
  
      """
      Unexpected exception or exit with exit_code !=0 on core worker.
      """
      a = Actor.remote()
      pid = ray.get(a.pid.remote())
      with pytest.raises(ray.exceptions.RayActorError):
          ray.get(a.exit.options(name="exit").remote(4))
  
      def verify_exit_failure():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          # If the worker is killed by SIGKILL, it is highly likely by OOM, so
          # the error message should contain information.
          assert type == "SYSTEM_ERROR" and "exit code 4" in detail
  
          # Verify the task failed with the info.
          return verify_failed_task(
              name="exit",
              error_type="ACTOR_DIED",
              error_message="exit code 4",
          )
  
      wait_for_condition(verify_exit_failure)
  
file_path: ray/python/ray/tests/test_exit_observability.py
imports: import asyncio
  import os
  import signal
  import sys
  import pytest
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.util.state import list_workers, list_nodes, list_tasks
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 2 for calling a global function with example usage: test_worker_exit_system_error(ray_start_cluster) and returns: unknown, verify_failed_task, os