code: def assign_partitions_to_actors(
      self, actors: List["ActorHandle"]
  ) -> Tuple[List[List[int]], List[str], List["ActorHandle"]]:
      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:
          invalidInputError(
              hasattr(actor, "get_node_ip"),  # type:ignore
              "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))
      ]  # type:ignore
  
      ip2actors = {}  # type:ignore
      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 __future__ import annotations
  from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from bigdl.orca.data import XShards
  from bigdl.orca.ray import OrcaRayContext
  import logging
  from bigdl.dllib.utils.log4Error import invalidInputError
  from numpy import ndarray
  from typing import TYPE_CHECKING, Any, Callable, DefaultDict, Dict, List, Tuple, Union, no_type_check
label_desc: Use image node_id 13 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