code: def test_no_scaledown_with_spilled_objects(shutdown_only):
      cluster = AutoscalingCluster(
          head_resources={"CPU": 0},
          worker_node_types={
              "cpu_node": {
                  "resources": {
                      "CPU": 1,
                      "object_store_memory": 75 * 1024 * 1024,
                  },
                  "node_config": {},
                  "min_workers": 0,
                  "max_workers": 2,
              },
          },
          idle_timeout_minutes=0.05,
      )
  
      try:
          cluster.start(
              _system_config={
                  "scheduler_report_pinned_bytes_only": True,
                  "min_spilling_size": 0,
              }
          )
          ray.init("auto")
  
          actors = [Actor.remote() for _ in range(2)]
          ray.get([a.f.remote() for a in actors])
  
          # Verify scale-up.
          wait_for_condition(
              lambda: ray.cluster_resources().get("CPU", 0) == 2
          )
          print("All nodes launched")
  
          # Put 10 x 80MiB objects into the object store with 75MiB memory limit.
          obj_size = 10 * 1024 * 1024
          objs = []
          for i in range(10):
              obj = actors[0].create.remote(obj_size)
              ray.get(actors[1].recv.remote(obj))
              objs.append(obj)
              print(f"obj {i}={obj.hex()}")
              del obj
  
          # At least 9 out of the 10 objects should have spilled.
          check_memory(
              [obj.hex() for obj in objs], num_spilled_objects=9
          )
          print(
              "Objects spilled, deleting actors and object references."
          )
  
          # Assume the 1st object always gets spilled.
          spilled_obj = objs[0]
          del objs
          del actors
  
          # Verify scale-down to 1 node.
          def scaledown_to_one():
              cpu = ray.cluster_resources().get("CPU", 0)
              assert cpu > 0, "Scale-down should keep at least 1 node"
              return cpu == 1
  
          wait_for_condition(scaledown_to_one, timeout=30)
  
          # Verify the spilled object still exists, and there is no object in the
          # plasma store.
          check_memory([spilled_obj.hex()], num_plasma_objects=0)
  
          # Delete the spilled object, the remaining worker node should be scaled
          # down.
          del spilled_obj
          wait_for_condition(
              lambda: ray.cluster_resources().get("CPU", 0) == 0
          )
          check_memory([], num_plasma_objects=0)
      finally:
          cluster.shutdown()
  
file_path: ray/python/ray/tests/test_autoscaler_fake_scaledown.py
imports: import platform
  import re
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.cluster_utils import AutoscalingCluster
label_desc: Use image node_id 3 for calling a global function with example usage: test_no_scaledown_with_spilled_objects(shutdown_only) and returns: unknown