presto-raptor-legacy/src/main/java/io/prestosql/plugin/raptor/legacy/storage/BucketBalancer.java/BucketBalancer/fetchClusterState
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());
    }