code: def test_critical_object_store_mem_resource_utilization(
      ray_start_cluster,
  ):
      cluster = ray_start_cluster
      cluster.add_node(
          _system_config={
              "scheduler_spread_threshold": 0.0,
          },
      )
      ray.init(address=cluster.address)
      non_local_node = cluster.add_node()
      cluster.wait_for_nodes()
  
      x = ray.put(np.zeros(1024 * 1024, dtype=np.uint8))
      print(x)
  
      @ray.remote
      def f():
          return ray._private.worker.global_worker.node.unique_id
  
      # Wait for resource availabilities to propagate.
      time.sleep(1)
      # The task should be scheduled to the remote node since
      # local node has non-zero object store mem utilization.
      assert ray.get(f.remote()) == non_local_node.unique_id
  
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 2 for calling a global function with example usage: test_critical_object_store_mem_resource_utilization(ray_start_cluster) and returns: ray