code: def test_task_args(ray_start_cluster):
      cluster = ray_start_cluster
      cluster.add_node(
          num_cpus=1,
          object_store_memory=80 * 1024 * 1024,
          _system_config={
              "local_fs_capacity_threshold": 0,
          },
          resources={"out_of_memory": 1},
      )
      cluster.add_node(
          num_cpus=1,
          object_store_memory=200 * 1024 * 1024,
          resources={"sufficient_memory": 1},
      )
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      @ray.remote
      def foo():
          return np.random.rand(20 * 1024 * 1024)  # 160 MB data
  
      @ray.remote
      def bar(obj):
          print(obj)
  
      ref = foo.options(resources={"sufficient_memory": 1}).remote()
      try:
          ray.get(
              bar.options(resources={"out_of_memory": 1}).remote(ref)
          )
      except ray.exceptions.RayTaskError as e:
          assert isinstance(e.cause, ray.exceptions.OutOfDiskError)
  
file_path: ray/python/ray/tests/test_out_of_disk_space.py
imports: import os
  import platform
  import shutil
  import sys
  import tempfile
  import time
  from contextlib import contextmanager
  import numpy
  import pytest
  import ray
  from ray.util.state import list_cluster_events
label_desc: Use image node_id 7 for calling a global function with example usage: test_task_args(ray_start_cluster) and returns: np