code: def _test_component_failed(cluster, component_type):
      """Kill a component on all worker nodes and check workload succeeds."""
  
      # Submit many tasks with many dependencies.
      @ray.remote
      def f(x):
          return x
  
      @ray.remote
      def g(*xs):
          return 1
  
      # Kill the component on all nodes except the head node as the tasks
      # execute. Do this in a loop while submitting tasks between each
      # component failure.
      time.sleep(0.1)
      worker_nodes = get_other_nodes(cluster)
      assert len(worker_nodes) > 0
      for node in worker_nodes:
          process = node.all_processes[component_type][0].process
          # Submit a round of tasks with many dependencies.
          x = 1
          for _ in range(1000):
              x = f.remote(x)
  
          xs = [g.remote(1)]
          for _ in range(100):
              xs.append(g.remote(*xs))
              xs.append(g.remote(1))
  
          # Kill a component on one of the nodes.
          process.terminate()
          time.sleep(1)
          process.kill()
          process.wait()
          assert not process.poll() is None
  
          # Make sure that we can still get the objects after the
          # executing tasks died.
          ray.get(x)
          ray.get(xs)
  
file_path: ray/python/ray/tests/test_component_failures_2.py
imports: import signal
  import sys
  import time
  import pytest
  import ray
  import ray._private.ray_constants
  from ray._private.test_utils import get_other_nodes, wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 2 for calling a global function with example usage: _test_component_failed(cluster, component_type) and returns: x, int