code: def test_owner_assign_when_put(ray_start_cluster, actor_resources):
      cluster_node_config = [
          {"num_cpus": 1, "resources": {f"node{i+1}": 10}}
          for i in range(3)
      ]
      cluster = ray_start_cluster
      for kwargs in cluster_node_config:
          cluster.add_node(**kwargs)
      ray.init(address=cluster.address)
  
      @ray.remote(resources=actor_resources["creator"], num_cpus=0)
      class Creator:
          def gen_object_ref(self, data="test", owner=None):
              return ray.put(data, _owner=owner)
  
      @ray.remote(resources=actor_resources["owner"], num_cpus=0)
      class Owner:
          def __init__(self):
              self.ref = None
  
          def set_object_ref(self, ref):
              self.ref = ref
  
          def warmup(self):
              return 0
  
      @ray.remote(resources=actor_resources["borrower"], num_cpus=0)
      class Borrower:
          def get_object(self, ref):
              return ray.get(ref)
  
      owner = Owner.remote()
      creator = Creator.remote()
      borrower = Borrower.remote()
  
      # Make sure the owner actor is alive.
      ray.get(owner.warmup.remote())
  
      object_ref = creator.gen_object_ref.remote(
          data="test1", owner=owner
      )
      # TODO(Catch-Bull): Ideally, deleting this line can also work normally,
      # cause driver keep a reference of the object. But, for now, it still
      # requires the owner to keep a reference of the object to make it
      # available.
      ray.get(owner.set_object_ref.remote(object_ref))
  
      ray.kill(creator)
      time.sleep(10)
  
      data = ray.get(borrower.get_object.remote(object_ref))
      assert data == "test1"
  
      ray.kill(owner)
      time.sleep(2)
      with pytest.raises(ray.exceptions.RayTaskError) as error:
          ray.get(borrower.get_object.remote(object_ref), timeout=2)
      assert isinstance(
          error.value.as_instanceof_cause(), OwnerDiedError
      )
  
file_path: ray/python/ray/tests/test_object_assign_owner.py
imports: import pytest
  import ray
  import time
  import numpy
  import os
  from ray._private.test_utils import skip_flaky_core_test_premerge
  from ray.exceptions import OwnerDiedError
label_desc: Use image node_id 2 for calling a global function with example usage: test_owner_assign_when_put(ray_start_cluster, actor_resources) and returns: ray, int, ray