code: def test_streaming_fault_tolerance(
      ray_start_10_cpus_shared, restore_data_context
  ):
      DataContext.get_current().new_execution_backend = True
      DataContext.get_current().use_streaming_executor = True
  
      class RandomExit:
          def __call__(self, x):
              import os
  
              if random.random() > 0.9:
                  print("force exit")
                  os._exit(1)
              return x
  
      # Test recover.
      base = ray.data.range(1000, parallelism=100)
      ds1 = base.map_batches(
          RandomExit,
          compute=ray.data.ActorPoolStrategy(size=4),
          max_task_retries=999,
      )
      ds1.take_all()
  
      # Test disabling fault tolerance.
      ds2 = base.map_batches(
          RandomExit,
          compute=ray.data.ActorPoolStrategy(size=4),
          max_restarts=0,
      )
      with pytest.raises(ray.exceptions.RayActorError):
          ds2.take_all()
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 17 for calling a global function with example usage: test_streaming_fault_tolerance(ray_start_10_cpus_shared, restore_data_context) and returns: x