code: def test_spill_dir_cleanup_on_raylet_start(
      fs_only_object_spilling_config,
  ):
      (
          object_spilling_config,
          temp_folder,
      ) = fs_only_object_spilling_config
      cluster = Cluster()
      cluster.add_node(
          num_cpus=0,
          object_store_memory=75 * 1024 * 1024,
          _system_config={
              "object_spilling_config": object_spilling_config
          },
      )
      ray.init(address=cluster.address)
      node2 = cluster.add_node(
          num_cpus=1, object_store_memory=75 * 1024 * 1024
      )
  
      # This task will run on node 2 because node 1 has no CPU resource
      @ray.remote(num_cpus=1)
      def run_workload():
          ids = []
          for _ in range(2):
              arr = np.random.rand(5 * 1024 * 1024)  # 40 MB
              ids.append(ray.put(arr))
          return ids
  
      ids = ray.get(run_workload.remote())
      assert not is_dir_empty(temp_folder)
  
      # Kill node 2
      cluster.remove_node(node2)
  
      # Verify that the spill folder is not empty
      assert not is_dir_empty(temp_folder)
  
      # Start a new node
      cluster.add_node(num_cpus=1, object_store_memory=75 * 1024 * 1024)
  
      # Verify that the spill folder is now cleaned up
      assert is_dir_empty(temp_folder)
  
      # We hold the object refs to prevent them from being deleted
      del ids
      ray.shutdown()
      cluster.shutdown()
  
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 10 for calling a global function with example usage: test_spill_dir_cleanup_on_raylet_start(fs_only_object_spilling_config) and returns: ids