code: def test_streaming_split_invalid_iterator(ray_start_10_cpus_shared):
      ds = ray.data.range(20, parallelism=20)
      (
          i1,
          i2,
      ) = ds.streaming_split(2, equal=True)
  
      @ray.remote
      def consume(x, times):
          i = 0
          for _ in range(times):
              for _ in x.iter_rows():
                  i += 1
          return i
  
      # InvalidIterator error from too many concurrent readers.
      with pytest.raises(ValueError):
          ray.get(
              [
                  consume.remote(i1, 4),
                  consume.remote(i2, 4),
                  consume.remote(i1, 4),
                  consume.remote(i2, 4),
              ]
          )
  
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 8 for calling a global function with example usage: test_streaming_split_invalid_iterator(ray_start_10_cpus_shared) and returns: i