code: def test_metrics_batch(shutdown_only):
      """Verify metrics_report_batch_size works correctly without data loss."""
      config_copy = copy.deepcopy(METRIC_CONFIG)
      config_copy["_system_config"].update(
          {"metrics_report_batch_size": 1}
      )
      info = ray.init(num_cpus=2, **config_copy)
  
      driver = """
  import ray
  import os
  import time
  
  ray.init("auto")
  
  @ray.remote
  class Phaser:
      def __init__(self):
          self.i = 0
  
      def inc(self):
          self.i += 1
          if self.i < 3:
              raise ValueError("First two tries will fail")
  
  phaser = Phaser.remote()
  
  @ray.remote(max_restarts=10, max_task_retries=10)
  class F:
      def f(self):
          try:
              ray.get(phaser.inc.remote())
          except Exception:
              print("RESTART")
              os._exit(1)
  
  f = F.remote()
  ray.get(f.f.remote())
  time.sleep(999)
  """
  
      proc = run_string_as_driver_nonblocking(driver)
      expected = {
          ("F.__init__", "FINISHED", "0"): 1.0,
          ("F.f", "FAILED", "0"): 1.0,
          ("F.f", "FAILED", "1"): 1.0,
          ("F.f", "FINISHED", "1"): 1.0,
          ("Phaser.__init__", "FINISHED", "0"): 1.0,
          ("Phaser.inc", "FINISHED", "0"): 1.0,
      }
      wait_for_condition(
          lambda: tasks_by_all(info) == expected,
          timeout=20,
          retry_interval_ms=500,
      )
      proc.kill()
  
imports: from collections import defaultdict
  import sys
  import os
  import copy
  import pytest
  import ray
  from ray._private.metrics_agent import RAY_WORKER_TIMEOUT_S
  from ray._private.test_utils import raw_metrics, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
label_desc: Use image node_id 19 for calling a global function with example usage: test_metrics_batch(shutdown_only) without return types