code: def test_kill_driver_clears_backlog(call_ray_start):
      driver = """
  import ray
  
  @ray.remote
  def f():
      import time
      time.sleep(300)
  
  refs = [f.remote() for _ in range(10000)]
  
  ray.get(refs)
    """
      proc = run_string_as_driver_nonblocking(driver)
      ctx = ray.init(address=call_ray_start)
  
      def get_backlog_and_pending():
          resources_batch = get_resource_usage(
              gcs_address=ctx.address_info["gcs_address"]
          )
          backlog = (
              resources_batch.resource_load_by_shape.resource_demands[
                  0
              ].backlog_size
              if resources_batch.resource_load_by_shape.resource_demands
              else 0
          )
  
          pending = 0
          demands = get_load_metrics_report(
              webui_url=ctx.address_info["webui_url"]
          )["resourceDemand"]
          for demand in demands:
              resource_dict, amount = demand
              if "CPU" in resource_dict:
                  pending = amount
  
          return pending, backlog
  
      def check_backlog(expect_backlog) -> bool:
          pending, backlog = get_backlog_and_pending()
          if expect_backlog:
              return pending > 0 and backlog > 0
          else:
              return pending == 0 and backlog == 0
  
      wait_for_condition(
          check_backlog,
          timeout=10,
          retry_interval_ms=1000,
          expect_backlog=True,
      )
  
      os.kill(proc.pid, 9)
  
      wait_for_condition(
          check_backlog,
          timeout=10,
          retry_interval_ms=1000,
          expect_backlog=False,
      )
  
file_path: ray/python/ray/tests/test_node_manager.py
imports: import ray
  from ray.util.state import list_workers
  from ray._private.test_utils import get_load_metrics_report, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition, get_resource_usage
  import pytest
  import os
  from ray.util.state import list_objects
  import subprocess
  from ray._private.utils import get_num_cpus
  import time
  import sys
label_desc: Use image node_id 2 for calling a global function with example usage: test_kill_driver_clears_backlog(call_ray_start) and returns: pending,  backlog, unknown, unknown