presto-raptor-legacy/src/main/java/io/prestosql/plugin/raptor/legacy/storage/BucketBalancer.java/BucketBalancer/updateAssignments
class BucketBalancer:    private int updateAssignments(Multimap<String, BucketAssignment> sourceToAllocationChanges)
    {
        // perform moves in decreasing order of source node total assigned buckets
        List<String> sourceNodes = sourceToAllocationChanges.asMap().entrySet().stream()
                .sorted((a, b) -> Integer.compare(b.getValue().size(), a.getValue().size()))
                .map(Map.Entry::getKey)
                .collect(toList());

        int moves = 0;
        for (String source : sourceNodes) {
            for (BucketAssignment reassignment : sourceToAllocationChanges.get(source)) {
                // todo: rate-limit new assignments
                shardManager.updateBucketAssignment(reassignment.getDistributionId(), reassignment.getBucketNumber(), reassignment.getNodeIdentifier());
                bucketsBalanced.update(1);
                moves++;
                log.info("Distribution %s: Moved bucket %s from %s to %s",
                        reassignment.getDistributionId(),
                        reassignment.getBucketNumber(),
                        source,
                        reassignment.getNodeIdentifier());
            }
        }

        return moves;
    }