code: def test_data_locality_spilled_objects(
      ray_start_cluster_enabled, fs_only_object_spilling_config
  ):
      cluster = ray_start_cluster_enabled
      object_spilling_config, _ = fs_only_object_spilling_config
      cluster.add_node(
          num_cpus=1,
          object_store_memory=100 * 1024 * 1024,
          _system_config={
              "min_spilling_size": 1,
              "object_spilling_config": object_spilling_config,
          },
      )
      ray.init(cluster.address)
      cluster.add_node(
          num_cpus=1,
          object_store_memory=100 * 1024 * 1024,
          resources={"remote": 1},
      )
  
      @ray.remote(resources={"remote": 1})
      def f():
          return (
              np.zeros(50 * 1024 * 1024, dtype=np.uint8),
              ray.runtime_context.get_runtime_context().get_node_id(),
          )
  
      @ray.remote
      def check_locality(x):
          _, node_id = x
          assert (
              node_id
              == ray.runtime_context.get_runtime_context().get_node_id()
          )
  
      # Check locality works when dependent task is already submitted by the time
      # the upstream task finishes.
      for _ in range(5):
          ray.get(check_locality.remote(f.remote()))
  
      # Check locality works when some objects were spilled.
      xs = [f.remote() for _ in range(5)]
      ray.wait(xs, num_returns=len(xs), fetch_local=False)
      for i, x in enumerate(xs):
          task = check_locality.remote(x)
          print(i, x, task)
          ray.get(task)
  
imports: import os
  import platform
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.experimental.internal_kv
  from ray._private.test_utils import make_global_state_accessor, wait_for_condition, get_metric_check_condition, MetricSamplePattern
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, PlacementGroupSchedulingStrategy
  from ray._private.test_utils import SignalActor
label_desc: Use image node_id 11 for calling a global function with example usage: test_data_locality_spilled_objects(ray_start_cluster_enabled, fs_only_object_spilling_config) and returns: np,  ray