def _rebalance_partition_replicas(
            self,
            partition,
            max_movement_count=None,
            max_movement_size=None,
    ):
        """Rebalance replication groups for given partition."""
        # 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
Rebalance replication groups for given partition.