ClusterBalancer implements Balancer { @Override public void balanceAll(String source, int rebalanceConcurrency, Optional<RateLimiter> rebalanceRateLimiter) { balance(0, 0.0f, rebalanceConcurrency, Optional.of(source), rebalanceRateLimiter); } ClusterBalancer(DistributedLogClientBuilder clientBuilder);   ClusterBalancer(DistributedLogClientBuilder clientBuilder,
                    Pair<DistributedLogClient, MonitorServiceClient> clientPair); @Override void balanceAll(String source,
                           int rebalanceConcurrency, /* unused */
                           Optional<RateLimiter> rebalanceRateLimiter); @Override void balance(int rebalanceWaterMark,
                        double rebalanceTolerancePercentage,
                        int rebalanceConcurrency, /* unused */
                        Optional<RateLimiter> rebalanceRateLimiter); void balance(int rebalanceWaterMark,
                        double rebalanceTolerancePercentage,
                        int rebalanceConcurrency,
                        Optional<String> source,
                        Optional<RateLimiter> rebalanceRateLimiter); @Override void close();  }
@Ignore @Test(timeout = 60000) public void testBalanceAll() throws Exception { String namePrefix = "clusterbalancer-balance-all-"; initStreams(namePrefix); writeStreams(namePrefix); validateStreams(namePrefix); Optional<RateLimiter> rateLimiter = Optional.absent(); Balancer balancer = new ClusterBalancer(client.dlClientBuilder, Pair.of((DistributedLogClient)client.dlClient, (MonitorServiceClient)client.dlClient)); logger.info("Rebalancing from 'unknown' target"); try { balancer.balanceAll("unknown", 10, rateLimiter); fail("Should fail on balanceAll from 'unknown' target."); } catch (IllegalArgumentException iae) { } validateStreams(namePrefix); logger.info("Rebalancing from 'unexisted' host"); String addr = DLSocketAddress.toString(DLSocketAddress.getSocketAddress(9999)); balancer.balanceAll(addr, 10, rateLimiter); validateStreams(namePrefix); addr = DLSocketAddress.toString(cluster.get(0).getAddress()); logger.info("Rebalancing from host {}.", addr); balancer.balanceAll(addr, 10, rateLimiter); checkStreams(0, cluster.get(0)); checkStreams(4, cluster.get(1)); checkStreams(3, cluster.get(2)); checkStreams(4, cluster.get(3)); checkStreams(4, cluster.get(4)); addr = DLSocketAddress.toString(cluster.get(2).getAddress()); logger.info("Rebalancing from host {}.", addr); balancer.balanceAll(addr, 10, rateLimiter); checkStreams(3, cluster.get(0)); checkStreams(4, cluster.get(1)); checkStreams(0, cluster.get(2)); checkStreams(4, cluster.get(3)); checkStreams(4, cluster.get(4)); logger.info("Rebalancing the cluster"); balancer.balance(0, 0.0f, 10, rateLimiter); for (int i = 0; i < 5; i++) { checkStreams(3, cluster.get(i)); } }
@Ignore @Test(timeout = 60000) public void testBalanceAll() throws Exception { String namePrefix = "clusterbalancer-balance-all-"; initStreams(namePrefix); writeStreams(namePrefix); validateStreams(namePrefix); Optional<RateLimiter> rateLimiter = Optional.absent(); Balancer balancer = new ClusterBalancer(client.dlClientBuilder, Pair.of((DistributedLogClient) client.dlClient, (MonitorServiceClient) client.dlClient)); logger.info("Rebalancing from 'unknown' target"); try { balancer.balanceAll("unknown", 10, rateLimiter); fail("Should fail on balanceAll from 'unknown' target."); } catch (IllegalArgumentException iae) { } validateStreams(namePrefix); logger.info("Rebalancing from 'unexisted' host"); String addr = DLSocketAddress.toString(DLSocketAddress.getSocketAddress(9999)); balancer.balanceAll(addr, 10, rateLimiter); validateStreams(namePrefix); addr = DLSocketAddress.toString(cluster.get(0).getAddress()); logger.info("Rebalancing from host {}.", addr); balancer.balanceAll(addr, 10, rateLimiter); checkStreams(0, cluster.get(0)); checkStreams(4, cluster.get(1)); checkStreams(3, cluster.get(2)); checkStreams(4, cluster.get(3)); checkStreams(4, cluster.get(4)); addr = DLSocketAddress.toString(cluster.get(2).getAddress()); logger.info("Rebalancing from host {}.", addr); balancer.balanceAll(addr, 10, rateLimiter); checkStreams(3, cluster.get(0)); checkStreams(4, cluster.get(1)); checkStreams(0, cluster.get(2)); checkStreams(4, cluster.get(3)); checkStreams(4, cluster.get(4)); logger.info("Rebalancing the cluster"); balancer.balance(0, 0.0f, 10, rateLimiter); for (int i = 0; i < 5; i++) { checkStreams(3, cluster.get(i)); } }