code: def test_backlog_report(shutdown_only):
      cluster = ray.init(
          num_cpus=1,
          _system_config={
              "max_pending_lease_requests_per_scheduling_category": 1
          },
      )
  
      global_state_accessor = make_global_state_accessor(cluster)
  
      @ray.remote(num_cpus=1)
      def foo(x):
          print(".")
          time.sleep(x)
          return None
  
      def backlog_size_set():
          message = global_state_accessor.get_all_resource_usage()
          if message is None:
              return False
  
          resource_usage = gcs_utils.ResourceUsageBatchData.FromString(
              message
          )
          aggregate_resource_load = (
              resource_usage.resource_load_by_shape.resource_demands
          )
          if len(aggregate_resource_load) == 1:
              backlog_size = aggregate_resource_load[0].backlog_size
              print(backlog_size)
              # Ideally we'd want to assert backlog_size == 8, but guaranteeing
              # the order the order that submissions will occur is too
              # hard/flaky.
              return backlog_size > 0
          return False
  
      # We want this first task to finish
      refs = [foo.remote(0.5)]
      # These tasks should all start _before_ the first one finishes.
      refs.extend([foo.remote(1000) for _ in range(9)])
      # Now there's 1 request running, 1 queued in the raylet, and 8 queued in
      # the worker backlog.
  
      ray.get(refs[0])
      # First request finishes, second request is now running, third lease
      # request is sent to the raylet with backlog=7
  
      wait_for_condition(backlog_size_set, timeout=2)
      global_state_accessor.disconnect()
  
imports: import os
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray._private.ray_constants
  from ray._raylet import GcsClient
  from ray.core.generated import autoscaler_pb2
  from ray._private.test_utils import convert_actor_state, make_global_state_accessor, wait_for_condition
label_desc: Use image node_id 12 for calling a global function with example usage: test_backlog_report(shutdown_only) and returns: None, False, False, unknown