code: def test_two_custom_resources(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(
          num_cpus=3,
          resources={"CustomResource1": 1, "CustomResource2": 2},
      )
      custom_resource_node = cluster.add_node(
          num_cpus=3,
          resources={"CustomResource1": 3, "CustomResource2": 4},
      )
      ray.init(address=cluster.address)
  
      @ray.remote
      def foo():
          # Sleep a while to emulate a slow operation. This is needed to make
          # sure tasks are scheduled to different nodes.
          time.sleep(0.1)
          return ray._private.worker.global_worker.node.unique_id
  
      # Make sure each node has at least one idle worker.
      wait_for_condition(
          lambda: len(set(ray.get([foo.remote() for _ in range(6)])))
          == 2
      )
  
      # Make sure the resource view is refreshed.
      time.sleep(1)
  
      @ray.remote(resources={"CustomResource1": 1})
      def f():
          time.sleep(0.001)
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(resources={"CustomResource2": 1})
      def g():
          time.sleep(0.001)
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(
          resources={"CustomResource1": 1, "CustomResource2": 3}
      )
      def h():
          time.sleep(0.001)
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(resources={"CustomResource1": 4})
      def j():
          time.sleep(0.001)
          return ray._private.worker.global_worker.node.unique_id
  
      @ray.remote(resources={"CustomResource3": 1})
      def k():
          time.sleep(0.001)
          return ray._private.worker.global_worker.node.unique_id
  
      # The f and g tasks should be scheduled on both raylets.
      assert len(set(ray.get([f.remote() for _ in range(500)]))) == 2
      assert len(set(ray.get([g.remote() for _ in range(500)]))) == 2
  
      # The h tasks should be scheduled only on the second raylet.
      raylet_ids = set(ray.get([h.remote() for _ in range(50)]))
      assert len(raylet_ids) == 1
      assert list(raylet_ids)[0] == custom_resource_node.unique_id
  
      # Make sure that tasks with unsatisfied custom resource requirements do
      # not get scheduled.
      ready_ids, remaining_ids = ray.wait(
          [j.remote(), k.remote()], timeout=0.5
      )
      assert ready_ids == []
  
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 9 for calling a global function with example usage: test_two_custom_resources(ray_start_cluster) and returns: ray, ray, ray, ray, ray, ray