code: def test_chaos_task_retry(set_kill_interval):
      # Chaos testing.
      @ray.remote(max_retries=-1)
      def task():
          a = ""
          for _ in range(100000):
              a = a + random.choice(string.ascii_letters)
          return
  
      @ray.remote(max_retries=-1)
      def invoke_nested_task():
          time.sleep(0.8)
          return ray.get(task.remote())
  
      # 50MB of return values.
      TOTAL_TASKS = 100
  
      pb = ProgressBar("Chaos test sanity check", TOTAL_TASKS)
      results = [
          invoke_nested_task.remote() for _ in range(TOTAL_TASKS)
      ]
      start = time.time()
      pb.block_until_complete(results)
      runtime_with_failure = time.time() - start
      print(
          f"Runtime when there are many failures: {runtime_with_failure}"
      )
      pb.close()
  
imports: import sys
  import random
  import string
  import ray
  import pytest
  import time
  from ray.experimental import shuffle
  from ray.tests.conftest import _ray_start_chaos_cluster
  from ray.data._internal.progress_bar import ProgressBar
  from ray.util.placement_group import placement_group
  from ray._private.test_utils import NodeKillerActor, get_log_message, get_and_run_resource_killer, WorkerKillerActor, wait_for_condition
  from ray.exceptions import RayTaskError, ObjectLostError
  from ray.util.state.common import ListApiOptions, StateResource
  from ray.util.state.api import StateApiClient, list_nodes
  from ray.cluster_utils import AutoscalingCluster
label_desc: Use image node_id 3 for calling a global function with example usage: test_chaos_task_retry(set_kill_interval) and returns: ray