code: def test_output_split_e2e(ray_start_10_cpus_shared):
      executor = StreamingExecutor(ExecutionOptions())
      inputs = make_ref_bundles([[x] for x in range(20)])
      o1 = InputDataBuffer(inputs)
      o2 = OutputSplitter(o1, 2, equal=True)
      it = executor.execute(o2)
  
      class Consume(threading.Thread):
          def __init__(self, idx):
              self.idx = idx
              self.out = []
              super().__init__()
  
          def run(self):
              while True:
                  try:
                      self.out.append(
                          it.get_next(output_split_idx=self.idx)
                      )
                  except Exception as e:
                      print(e)
                      raise
  
      c0 = Consume(0)
      c1 = Consume(1)
      c0.start()
      c1.start()
      c0.join()
      c1.join()
  
      def get_outputs(out: List[RefBundle]):
          outputs = []
          for bundle in out:
              for block, _ in bundle.blocks:
                  ids: pd.Series = ray.get(block)["id"]
                  outputs.extend(ids.values)
          return outputs
  
      assert get_outputs(c0.out) == list(range(0, 20, 2))
      assert get_outputs(c1.out) == list(range(1, 20, 2))
      assert len(c0.out) == 10, c0.out
      assert len(c1.out) == 10, c0.out
  
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 5 for calling a global function with example usage: test_output_split_e2e(ray_start_10_cpus_shared) and returns: outputs