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
def _rebalance_partition_replicas ( self , partition , max_movement_count = None , max_movement_size = None , ) : 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 , ) 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 ) : 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 : 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 : break 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.