code: def test_non_corrupted_resources():
      """
      Test that when node's local gc happens due to object store pressure,
      the message doesn't corrupt the resource view on the gcs.
      See issue https://github.com/ray-project/ray/issues/39644
      """
      num_worker_nodes = 5
      cluster = AutoscalingCluster(
          head_resources={
              "CPU": 2,
              "object_store_memory": 100 * 1024 * 1024,
          },
          worker_node_types={
              "type-1": {
                  "resources": {"CPU": 2},
                  "node_config": {},
                  "min_workers": num_worker_nodes,
                  "max_workers": num_worker_nodes,
              },
          },
      )
  
      driver_script = """
  
  import ray
  import time
  
  ray.init("auto")
  
  @ray.remote(num_cpus=1)
  def foo():
      ray.put(bytearray(1024*1024* 50))
  
  
  while True:
      ray.get([foo.remote() for _ in range(50)])
  """
  
      try:
          # This should trigger many COMMANDS messages from NodeManager.
          cluster.start(
              _system_config={
                  "debug_dump_period_milliseconds": 10,
                  "raylet_report_resources_period_milliseconds": 10000,
                  "global_gc_min_interval_s": 1,
                  "local_gc_interval_s": 1,
                  "high_plasma_storage_usage": 0.2,
                  "raylet_check_gc_period_milliseconds": 10,
              },
          )
          ctx = ray.init("auto")
          gcs_address = ctx.address_info["gcs_address"]
  
          from ray.autoscaler.v2.sdk import get_cluster_status
  
          def nodes_up():
              cluster_state = get_cluster_status(gcs_address)
              return (
                  len(cluster_state.idle_nodes) == num_worker_nodes + 1
              )
  
          wait_for_condition(nodes_up)
  
          # Schedule tasks
          run_string_as_driver_nonblocking(driver_script)
          start = time.time()
  
          # Check the cluster state for 10 seconds
          while time.time() - start < 10:
              cluster_state = get_cluster_status(gcs_address)
  
              # Verify total cluster resources never change
              assert (
                  len(cluster_state.idle_nodes)
                  + len(cluster_state.active_nodes)
              ) == num_worker_nodes + 1
              assert cluster_state.total_resources()["CPU"] == 2 * (
                  num_worker_nodes + 1
              )
  
      finally:
          ray.shutdown()
          cluster.shutdown()
  
file_path: ray/python/ray/autoscaler/v2/tests/test_e2e.py
imports: import os
  import sys
  import time
  from typing import Dict
  import pytest
  import ray
  from ray._private.resource_spec import HEAD_NODE_RESOURCE_NAME
  from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
  from ray.autoscaler.v2.sdk import get_cluster_status
  from ray.cluster_utils import AutoscalingCluster
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.state.api import list_placement_groups, list_tasks
label_desc: Use image node_id 8 for calling a global function with example usage: test_non_corrupted_resources() and returns: unknown