code: def test_pull_request_retry(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(
          num_cpus=0, num_gpus=1, object_store_memory=100 * 2**20
      )
      cluster.add_node(
          num_cpus=1, num_gpus=0, object_store_memory=100 * 2**20
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      @ray.remote
      def put():
          return np.zeros(64 * 2**20, dtype=np.int8)
  
      @ray.remote(num_cpus=0, num_gpus=1)
      def driver():
          local_ref = ray.put(np.zeros(64 * 2**20, dtype=np.int8))
  
          remote_ref = put.remote()
  
          ready, _ = ray.wait([remote_ref], timeout=30)
          assert len(ready) == 1
  
          del local_ref
  
          # This should always complete within 10 seconds.
          ready, _ = ray.wait([remote_ref], timeout=20)
          assert len(ready) > 0
  
      # Pretend the GPU node is the driver. We do this to force the placement of
      # the driver and `put` task on different nodes.
      ray.get(driver.remote())
  
imports: import multiprocessing
  import time
  import warnings
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 7 for calling a global function with example usage: test_pull_request_retry(ray_start_cluster) and returns: np