code: def test_node_affinity_scheduling_strategy_spill_on_unavailable(
      ray_start_cluster,
  ):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=3)
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=3)
      cluster.wait_for_nodes()
  
      @ray.remote
      def get_node_id_task(sleep_s=0):
          time.sleep(sleep_s)
          return ray.get_runtime_context().get_node_id()
  
      target_node_id = ray.get(get_node_id_task.remote())
  
      _ = [
          get_node_id_task.options(
              scheduling_strategy=NodeAffinitySchedulingStrategy(
                  target_node_id, soft=False
              )
          ).remote(1000)
          for _ in range(3)
      ]
  
      soft_ref = get_node_id_task.options(
          scheduling_strategy=NodeAffinitySchedulingStrategy(
              target_node_id, soft=True, _spill_on_unavailable=True
          )
      ).remote()
  
      soft_node_id = ray.get(soft_ref, timeout=3)
      assert target_node_id != soft_node_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 6 for calling a global function with example usage: test_node_affinity_scheduling_strategy_spill_on_unavailable(ray_start_cluster) and returns: ray