code: def _rebalance_partition_replicas(
              self,
              partition,
              max_movement_count=None,
              max_movement_size=None,
      ):
          
          # Separate replication-groups into under and over replicated
          total = partition.replication_factor
          over_replicated_rgs, under_replicated_rgs = separate_groups(
              list(self.cluster_topology.rgs.values()),
              lambda g: g.count_replica(partition),
              total,
          )
  
          # Move replicas from over-replicated to under-replicated groups
          movement_count = 0
          movement_size = 0
          while (
              under_replicated_rgs and over_replicated_rgs
          ) and (
              max_movement_size is None or
              movement_size + partition.size <= max_movement_size
          ) and (
              max_movement_count is None or
              movement_count < max_movement_count
          ):
              # Decide source and destination group
              rg_source = self._elect_source_replication_group(
                  over_replicated_rgs,
                  partition,
              )
              rg_destination = self._elect_dest_replication_group(
                  rg_source.count_replica(partition),
                  under_replicated_rgs,
                  partition,
              )
              if rg_source and rg_destination:
                  # Actual movement of partition
                  self.log.debug(
                      'Moving partition {p_name} from replication-group '
                      '{rg_source} to replication-group {rg_dest}'.format(
                          p_name=partition.name,
                          rg_source=rg_source.id,
                          rg_dest=rg_destination.id,
                      ),
                  )
                  rg_source.move_partition(rg_destination, partition)
                  movement_count += 1
                  movement_size += partition.size
              else:
                  # Groups balanced or cannot be balanced further
                  break
              # Re-compute under and over-replicated replication-groups
              over_replicated_rgs, under_replicated_rgs = separate_groups(
                  list(self.cluster_topology.rgs.values()),
                  lambda g: g.count_replica(partition),
                  total,
              )
          return movement_count, movement_size
docs: Rebalance replication groups for given partition.