code: def test_random_shuffle_spread(
      ray_start_cluster, use_push_based_shuffle
  ):
      cluster = ray_start_cluster
      cluster.add_node(
          resources={"bar:1": 100},
          num_cpus=10,
          _system_config={"max_direct_call_object_size": 0},
      )
      cluster.add_node(resources={"bar:2": 100}, num_cpus=10)
      cluster.add_node(resources={"bar:3": 100}, num_cpus=0)
  
      ray.init(cluster.address)
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      node1_id = ray.get(
          get_node_id.options(resources={"bar:1": 1}).remote()
      )
      node2_id = ray.get(
          get_node_id.options(resources={"bar:2": 1}).remote()
      )
  
      ds = ray.data.range(100, parallelism=2).random_shuffle()
      blocks = ds.get_internal_block_refs()
      ray.wait(blocks, num_returns=len(blocks), fetch_local=False)
      location_data = ray.experimental.get_object_locations(blocks)
      locations = []
      for block in blocks:
          locations.extend(location_data[block]["node_ids"])
      assert "2 nodes used" in ds.stats()
  
      if not use_push_based_shuffle:
          # We don't check this for push-based shuffle since it will try to
          # colocate reduce tasks to improve locality.
          assert set(locations) == {node1_id, node2_id}
  
imports: import itertools
  import math
  import random
  import time
  from unittest.mock import patch
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data.aggregate import AggregateFn, Count, Max, Mean, Min, Quantile, Std, Sum
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, named_values
  from ray.tests.conftest import *
label_desc: Use image node_id 42 for calling a global function with example usage: test_random_shuffle_spread(ray_start_cluster, use_push_based_shuffle) and returns: ray