code: def test_serialized_id(ray_start_cluster):
      @ray.remote
      def small_object():
          # Sleep a bit before creating the object to force a timeout
          # at the getter.
          time.sleep(1)
          return 1
  
      @ray.remote
      def dependent_task(x):
          return x
  
      @ray.remote
      def get(obj_refs, test_dependent_task):
          print("get", obj_refs)
          obj_ref = obj_refs[0]
          if test_dependent_task:
              assert ray.get(dependent_task.remote(obj_ref)) == 1
          else:
              assert ray.get(obj_ref) == 1
  
      obj = small_object.remote()
      ray.get(get.remote([obj], False))
  
      obj = small_object.remote()
      ray.get(get.remote([obj], True))
  
      obj = ray.put(1)
      ray.get(get.remote([obj], False))
  
      obj = ray.put(1)
      ray.get(get.remote([obj], True))
  
imports: import os
  import signal
  import sys
  import threading
  import time
  import numpy
  import pytest
  import ray
  import ray._private.ray_constants
  import ray._private.utils
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR
  from ray._private.test_utils import Semaphore, get_error_message, get_log_batch, init_error_pubsub, run_string_as_driver_nonblocking, wait_for_condition
  from ray.cluster_utils import cluster_not_supported
  from ray.experimental.internal_kv import _internal_kv_get
label_desc: Use image node_id 7 for calling a global function with example usage: test_serialized_id(ray_start_cluster) and returns: int, x