code: def test_split_operator_locality_hints(ray_start_regular_shared):
      input_op = InputDataBuffer(
          make_ref_bundles([[i] for i in range(10)])
      )
      op = OutputSplitter(
          input_op, 2, equal=False, locality_hints=["node1", "node2"]
      )
  
      def get_fake_loc(item):
          assert isinstance(item, int), item
          if item in [0, 1, 4, 5, 8]:
              return "node1"
          else:
              return "node2"
  
      def get_bundle_loc(bundle):
          block = ray.get(bundle.blocks[0][0])
          fval = list(block["id"])[0]
          return get_fake_loc(fval)
  
      op._get_location = get_bundle_loc
  
      # Feed data and implement streaming exec.
      output_splits = collections.defaultdict(list)
      op.start(ExecutionOptions())
      while input_op.has_next():
          op.add_input(input_op.get_next(), 0)
      op.all_inputs_done()
      while op.has_next():
          ref = op.get_next()
          assert ref.owns_blocks, ref
          for block, _ in ref.blocks:
              output_splits[ref.output_split_idx].extend(
                  list(ray.get(block)["id"])
              )
  
      total = 0
      for i in range(2):
          if i == 0:
              node = "node1"
          else:
              node = "node2"
          split = output_splits[i]
          for item in split:
              assert get_fake_loc(item) == node
              total += 1
  
      assert total == 10, total
      assert "all objects local" in op.progress_str()
  
imports: import collections
  import random
  import time
  from typing import Any, Iterable, List
  from unittest.mock import MagicMock
  import numpy
  import pandas
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.compute import ActorPoolStrategy, TaskPoolStrategy
  from ray.data._internal.execution.interfaces import ExecutionOptions, PhysicalOperator, RefBundle
  from ray.data._internal.execution.operators.actor_pool_map_operator import ActorPoolMapOperator
  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.limit_operator import LimitOperator
  from ray.data._internal.execution.operators.map_operator import MapOperator, _BlockRefBundler
  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.operators.task_pool_map_operator import TaskPoolMapOperator
  from ray.data._internal.execution.operators.union_operator import UnionOperator
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.tests.util import run_one_op_task, run_op_tasks_sync
  from ray.tests.conftest import *
label_desc: Use image node_id 11 for calling a global function with example usage: test_split_operator_locality_hints(ray_start_regular_shared) and returns: get_fake_loc, str, str