### Question:
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  BucketBalancer(
            NodeManager nodeManager,
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            BucketBalancerConfig config,
            BackupService backupService,
            RaptorConnectorId connectorId);  BucketBalancer(
            NodeSupplier nodeSupplier,
            ShardManager shardManager,
            boolean enabled,
            Duration interval,
            boolean backupAvailable,
            boolean coordinator,
            String connectorId); @PostConstruct void start(); @PreDestroy void shutdown(); @Managed @Nested CounterStat getBucketsBalanced(); @Managed @Nested CounterStat getJobErrors(); @Managed void startBalanceJob();  }### Answer:
@Test public void testMultipleDistributionUnbalancedWithDiskSpace() { long distributionA = createDistribution("distA", 4); createBucketedTable("testA", distributionA, DataSize.valueOf("4B")); createAssignments(distributionA, AVAILABLE_WORKERS, 1, 1, 1, 1, 0); long distributionB = createDistribution("distB", 4); createBucketedTable("testB", distributionB, DataSize.valueOf("4B")); createAssignments(distributionB, AVAILABLE_WORKERS, 1, 1, 1, 0, 1); long distributionC = createDistribution("distC", 2); createBucketedTable("testC", distributionC, DataSize.valueOf("2B")); createAssignments(distributionC, AVAILABLE_WORKERS, 0, 0, 0, 2, 0); assertBalancing(balancer, 1); assertEquals(balancer.fetchClusterState().getAssignedBytes().values() .stream() .distinct() .count(), 1); }