code: def test_pull_spilled_object_failure(
      object_spilling_config, ray_start_cluster
  ):
      object_spilling_config, temp_folder = object_spilling_config
      cluster = ray_start_cluster
  
      # Head node.
      cluster.add_node(
          num_cpus=1,
          resources={"custom": 0},
          object_store_memory=75 * 1024 * 1024,
          _system_config={
              "max_io_workers": 2,
              "min_spilling_size": 1 * 1024 * 1024,
              "automatic_object_spilling_enabled": True,
              "object_store_full_delay_ms": 100,
              "object_spilling_config": object_spilling_config,
          },
      )
      ray.init(cluster.address)
  
      # add 1 worker node
      cluster.add_node(
          num_cpus=1,
          resources={"custom": 1},
          object_store_memory=75 * 1024 * 1024,
      )
      cluster.wait_for_nodes()
  
      @ray.remote(num_cpus=1, resources={"custom": 1})
      def create_objects():
          arr = np.random.rand(5 * 1024 * 1024)
          hash_value = zlib.crc32(arr.tobytes())
          results = [ray.put(arr), hash_value]
          # ensure the objects are spilled
          arr = np.random.rand(5 * 1024 * 1024)
          ray.get(ray.put(arr))
          ray.get(ray.put(arr))
          return results
  
      @ray.remote(num_cpus=1, resources={"custom": 0})
      def get_object(arr):
          return zlib.crc32(arr.tobytes())
  
      [ref, hash_value] = ray.get(create_objects.remote())
  
      # remove spilled file
      shutil.rmtree(temp_folder)
  
      hash_value1 = ray.get(get_object.remote(ref))
      assert hash_value == hash_value1
  
imports: import json
  import platform
  import random
  import re
  import shutil
  import sys
  import time
  import zlib
  from collections import defaultdict
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
  from ray.tests.test_object_spilling import assert_no_thrashing, is_dir_empty
label_desc: Use image node_id 9 for calling a global function with example usage: test_pull_spilled_object_failure(object_spilling_config, ray_start_cluster) and returns: results, zlib