class BucketBalancer {

    @VisibleForTesting
    ClusterState fetchClusterState()
    {
        Set<String> activeNodes = nodeSupplier.getWorkerNodes().stream()
                .map(Node::getNodeIdentifier)
                .collect(toSet());

        Map<String, Long> assignedNodeSize = new HashMap<>(activeNodes.stream().collect(toMap(node -> node, node -> 0L)));
        ImmutableMultimap.Builder<Distribution, BucketAssignment> distributionAssignments = ImmutableMultimap.builder();
        ImmutableMap.Builder<Distribution, Long> distributionBucketSize = ImmutableMap.builder();

        for (Distribution distribution : shardManager.getDistributions()) {
            long distributionSize = shardManager.getDistributionSizeInBytes(distribution.getId());
            long bucketSize = (long) (1.0 * distributionSize) / distribution.getBucketCount();
            distributionBucketSize.put(distribution, bucketSize);

            for (BucketNode bucketNode : shardManager.getBucketNodes(distribution.getId())) {
                String node = bucketNode.getNodeIdentifier();
                distributionAssignments.put(distribution, new BucketAssignment(distribution.getId(), bucketNode.getBucketNumber(), node));
                assignedNodeSize.merge(node, bucketSize, Math::addExact);
            }
        }

        return new ClusterState(activeNodes, assignedNodeSize, distributionAssignments.build(), distributionBucketSize.build());
    }

    @Inject public  BucketBalancer(
            NodeManager nodeManager,
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            BucketBalancerConfig config,
            BackupService backupService,
            RaptorConnectorId connectorId);
    public  BucketBalancer(
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            boolean enabled,
            Duration interval,
            boolean backupAvailable,
            boolean coordinator,
            String connectorId);

}