code: def test_spillback_waiting_task_on_oom(ray_start_cluster):
      # This test ensures that tasks are spilled if they are not schedulable due
      # to lack of object store memory.
      cluster = ray_start_cluster
      object_size = 1e8
      cluster.add_node(
          num_cpus=1,
          memory=1e9,
          object_store_memory=object_size * 2,
          _system_config={
              "automatic_object_spilling_enabled": False,
              "locality_aware_leasing_enabled": False,
          },
      )
      ray.init(address=cluster.address)
      cluster.add_node(
          num_cpus=1,
          resources={"custom": 1},
          memory=1e9,
          object_store_memory=object_size * 2,
      )
  
      @ray.remote(resources={"custom": 1})
      def create_remote_object():
          return np.zeros(int(object_size), dtype=np.uint8)
  
      local_obj = ray.put(
          np.zeros(int(object_size * 1.5), dtype=np.uint8)
      )
      print(local_obj)
  
      @ray.remote
      def f(x):
          return
  
      dep = create_remote_object.remote()
      ray.wait([dep], fetch_local=False)
      # Wait for resource availabilities to propagate.
      time.sleep(1)
      # This task can't run on the local node. Make sure it gets spilled even
      # though we have the local CPUs to run it.
      ray.get(f.remote(dep), timeout=30)
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 7 for calling a global function with example usage: test_spillback_waiting_task_on_oom(ray_start_cluster) and returns: np