code: def test_lease_request_leak(shutdown_only):
      ray.init(
          num_cpus=1,
          _system_config={"object_timeout_milliseconds": 200},
      )
  
      @ray.remote
      def f(x):
          time.sleep(0.1)
          return
  
      # Submit pairs of tasks. Tasks in a pair can reuse the same worker leased
      # from the raylet.
      tasks = []
      for _ in range(10):
          obj_ref = ray.put(1)
          for _ in range(2):
              tasks.append(f.remote(obj_ref))
          del obj_ref
      ray.get(tasks)
  
      wait_for_condition(lambda: object_memory_usage() == 0)
  
imports: import collections
  import logging
  import platform
  import subprocess
  import sys
  import time
  import unittest
  import numpy
  import pytest
  import ray
  import ray.cluster_utils
  import ray.util.accelerators
  from ray._private.internal_api import memory_summary
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray._private.test_utils import Semaphore, SignalActor, object_memory_usage, get_metric_check_condition, wait_for_condition, MetricSamplePattern
label_desc: Use image node_id 12 for calling a global function with example usage: test_lease_request_leak(shutdown_only) without return types