code: def assign_partitions_to_actors(self, actors):
      num_parts = self.num_partitions()
      if num_parts < len(actors):
          logger.warning(
              f"this rdd has {num_parts} partitions, which is smaller "
              f"than actor number ({len(actors)} actors). That could cause "
              f"unbalancing workload on different actors. We recommend you to "
              f"repartition the rdd for better performance."
          )
  
      avg_part_num = num_parts // len(actors)
      remainder = num_parts % len(actors)
  
      part_id2ip = self.partition2ip.copy()
      # the assigning algorithm
      # 1. calculate the average partition number per actor avg_part_num, and the number
      #    of remaining partitions remainder. So there are remainder number of actors got
      #    avg_part_num + 1 partitions and other actors got avg_part_num partitions.
      # 2. loop partitions and assign each according to ip, if no actor with this ip or
      #    all actors with this ip have been full, this round of assignment failed.
      # 3. assign the partitions that failed to be assigned to actors that has full
  
      # todo extract this algorithm to other functions for unit tests.
      actor_ips = []
      for actor in actors:
          assert hasattr(
              actor, "get_node_ip"
          ), "each actor should have a get_node_ip method"
          actor_ip = actor.get_node_ip.remote()
          actor_ips.append(actor_ip)
  
      actor_ips = ray.get(actor_ips)
  
      actor2assignments = [[] for i in range(len(actors))]
  
      ip2actors = {}
      for idx, ip in enumerate(actor_ips):
          if ip not in ip2actors:
              ip2actors[ip] = []
          ip2actors[ip].append(idx)
  
      unassigned = []
      for part_idx, ip in part_id2ip.items():
          assigned = False
          if ip in ip2actors:
              ip_actors = ip2actors[ip]
  
              for actor_id in ip_actors:
                  current_assignments = actor2assignments[actor_id]
                  if len(current_assignments) < avg_part_num:
                      current_assignments.append(part_idx)
                      assigned = True
                      break
                  elif (
                      len(current_assignments) == avg_part_num
                      and remainder > 0
                  ):
                      current_assignments.append(part_idx)
                      remainder -= 1
                      assigned = True
                      break
          if not assigned:
              unassigned.append((part_idx, ip))
  
      for part_idx, ip in unassigned:
          for current_assignments in actor2assignments:
              if len(current_assignments) < avg_part_num:
                  current_assignments.append(part_idx)
                  break
              elif (
                  len(current_assignments) == avg_part_num
                  and remainder > 0
              ):
                  current_assignments.append(part_idx)
                  remainder -= 1
                  break
  
      if num_parts < len(actors):
          # filter assigned actors
          assigned_actors = []
          assigned_actor2assignments = []
          assigned_actor_ips = []
          for actor, assignment, ip in zip(
              actors, actor2assignments, actor_ips
          ):
              if assignment:
                  assigned_actors.append(actor)
                  assigned_actor2assignments.append(assignment)
                  assigned_actor_ips.append(ip)
          return (
              assigned_actor2assignments,
              assigned_actor_ips,
              assigned_actors,
          )
      else:
          return actor2assignments, actor_ips, actors
  
imports: from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from packaging import version
  from zoo.orca.data import XShards
  from zoo.ray import RayContext
  import logging
label_desc: Use image node_id 11 for calling the RayXShards obj's underlying member method code with example usage: obj.assign_partitions_to_actors(actors) and returns: assigned_actor2assignments,  assigned_actor_ips,  assigned_actors, actor2assignments,  actor_ips,  actors