code: def test_maximize_concurrent_pull_race_condition(
      ray_start_cluster_head,
  ):
      # Test if https://github.com/ray-project/ray/issues/18062 is mitigated
      cluster = ray_start_cluster_head
      cluster.add_node(num_cpus=8, object_store_memory=75 * 1024 * 1024)
  
      @ray.remote
      class RemoteObjectCreator:
          def put(self, i):
              return np.random.rand(i * 1024 * 1024)  # 8 MB data
  
          def idle(self):
              pass
  
      @ray.remote
      def f(x):
          print(f"timestamp={time.time()} pulled {len(x)*8} bytes")
          time.sleep(1)
          return
  
      remote_obj_creator = RemoteObjectCreator.remote()
      remote_refs = [remote_obj_creator.put.remote(1) for _ in range(7)]
      print(remote_refs)
      # Make sure all objects are created.
      ray.get(remote_obj_creator.idle.remote())
  
      local_refs = [
          ray.put(np.random.rand(1 * 1024 * 1024)) for _ in range(20)
      ]
      remote_tasks = [f.remote(x) for x in local_refs]
  
      start = time.time()
      ray.get(remote_tasks)
      end = time.time()
      assert (
          end - start < 20
      ), "Too much time spent in pulling objects, check the amount of time in retries"
  
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 16 for calling a global function with example usage: test_maximize_concurrent_pull_race_condition(ray_start_cluster_head) and returns: np