code: def test_restartable_actor_throws_oom_error(
      ray_with_memory_monitor,
  ):
      addr = ray_with_memory_monitor
      leaker = Leaker.options(
          max_restarts=1, max_task_retries=1
      ).remote()
  
      bytes_to_alloc = get_additional_bytes_to_reach_memory_usage_pct(
          memory_usage_threshold + 0.1
      )
      with pytest.raises(ray.exceptions.OutOfMemoryError) as _:
          ray.get(
              leaker.allocate.remote(
                  bytes_to_alloc, memory_monitor_refresh_ms * 3
              )
          )
  
      wait_for_condition(
          has_metric_tagged_with_value,
          timeout=10,
          retry_interval_ms=100,
          addr=addr,
          tag="MemoryManager.ActorEviction.Total",
          value=2.0,
      )
  
      wait_for_condition(
          has_metric_tagged_with_value,
          timeout=10,
          retry_interval_ms=100,
          addr=addr,
          tag="Leaker.__init__",
          value=2.0,
      )
  
imports: from math import ceil
  import sys
  import time
  import pytest
  import ray
  from ray._private import ray_constants
  import ray._private.gcs_utils
  from ray._private.test_utils import wait_for_condition, raw_metrics
  import numpy
  from ray._private.utils import get_system_memory
  from ray._private.utils import get_used_memory
  from ray._private.state_api_test_utils import verify_failed_task
  from ray.util.state.state_manager import StateDataSourceClient
label_desc: Use image node_id 8 for calling a global function with example usage: test_restartable_actor_throws_oom_error(ray_with_memory_monitor) without return types