BalancerUtils { public static <K> int calculateNumStreamsToRebalance(K nodeToRebalance, Map<K, Integer> loadDistribution, int rebalanceWaterMark, double tolerancePercentage) { Integer myLoad = loadDistribution.get(nodeToRebalance); if (null == myLoad || myLoad <= rebalanceWaterMark) { return 0; } long totalLoad = 0L; int numNodes = loadDistribution.size(); for (Map.Entry<K, Integer> entry : loadDistribution.entrySet()) { if (null == entry.getKey() || null == entry.getValue()) { continue; } totalLoad += entry.getValue(); } double averageLoad = ((double) totalLoad) / numNodes; long permissibleLoad = Math.max(1L, (long) Math.ceil(averageLoad + averageLoad * tolerancePercentage / 100.0f)); if (myLoad <= permissibleLoad) { return 0; } return Math.max(0, myLoad - (int) Math.ceil(averageLoad)); }  static int calculateNumStreamsToRebalance(K nodeToRebalance,
                                                         Map<K, Integer> loadDistribution,
                                                         int rebalanceWaterMark,
                                                         double tolerancePercentage);  }
@Test(timeout = 60000) public void testCalculateNumStreamsToRebalance() { String myNode = "mynode"; assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance( myNode, new HashMap<String, Integer>(), 0, 10)); Map<String, Integer> loadDistribution = new HashMap<String, Integer>(); loadDistribution.put("node1", 10); assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance( myNode, loadDistribution, 0, 10)); loadDistribution.clear(); loadDistribution.put("node1", 1); loadDistribution.put(myNode, 100); assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance( myNode, loadDistribution, 200, 10)); loadDistribution.clear(); loadDistribution.put(myNode, 1); loadDistribution.put("node1", 99); assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance( myNode, loadDistribution, 0, 10)); assertEquals(49, BalancerUtils.calculateNumStreamsToRebalance( "node1", loadDistribution, 0, 10)); loadDistribution.clear(); loadDistribution.put(myNode, 55); loadDistribution.put("node1", 45); assertEquals(0, BalancerUtils.calculateNumStreamsToRebalance( myNode, loadDistribution, 0, 10)); }