code: def test_custom_resources(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1, resources={"CustomResource": 0})
      custom_resource_node = cluster.add_node(
          num_cpus=1, resources={"CustomResource": 1}
      )
      ray.init(address=cluster.address)
  
      @ray.remote
      def f():
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(resources={"CustomResource": 1})
      def g():
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(resources={"CustomResource": 1})
      def h():
          ray.get([f.remote() for _ in range(5)])
          return ray._private.worker.global_worker.node.unique_id
  
      # The g tasks should be scheduled only on the second raylet.
      raylet_ids = set(ray.get([g.remote() for _ in range(50)]))
      assert len(raylet_ids) == 1
      assert list(raylet_ids)[0] == custom_resource_node.unique_id
  
      # Make sure that resource bookkeeping works when a task that uses a
      # custom resources gets blocked.
      ray.get([h.remote() for _ in range(5)])
  
imports: import logging
  import os
  import sys
  import time
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  from ray._private.test_utils import RayTestTimeoutException, wait_for_condition
  from ray.util.placement_group import placement_group
  from ray.util.accelerators import AWS_NEURON_CORE
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 7 for calling a global function with example usage: test_custom_resources(ray_start_cluster) and returns: ray, ray, ray