StaticCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { int count = 0; int toTake = getToTakePartitions(topic).size(); if (toTake > 0) { count = toTake; } else { int toRelease = getToReleasePartitions(topic, -1).size(); if (toRelease > 0) count = (-toRelease); } countMap.put(topic, count); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } StaticCoordinator(String config, List<String> topics,
			String groupId, String consumerId, KafkaController.ZkConnector zk); @Override synchronized Map<String, Integer> getRebalanceCount(); @Override synchronized Set<Integer> getToTakePartitions(String topic); @Override synchronized Set<Integer> getToReleasePartitions(String topic,
			int count);  }
@Test public void testGetRebalanceCount() throws Exception { String groupId1 = "test_group1"; String topic0 = "Topic.test-0"; String topic1 = "Topic.test-1"; List<String> topics1 = new ArrayList<String>(); topics1.add(topic0); topics1.add(topic1); String consumerId0 = "D-00456-consumerId-0"; String consumerId1 = "D-00456-consumerId-1"; String path0 = "/brokers/topics/" + topic0 + "/partitions"; zkConnector.create(path0, false); zkConnector.create(path0 + "/0", false); zkConnector.create(path0 + "/1", false); zkConnector.create(path0 + "/2", false); zkConnector.create(path0 + "/3", false); zkConnector.create(path0 + "/4", false); String path1 = "/brokers/topics/" + topic1 + "/partitions"; zkConnector.create(path1, false); zkConnector.create(path1 + "/0", false); zkConnector.create(path1 + "/1", false); zkConnector.create(path1 + "/2", false); zkConnector.create(path1 + "/3", false); zkConnector.create(path1 + "/4", false); ZkCoordinator zkCoordinator0 = new StaticCoordinator("a,-1,1,2,3", topics1, groupId1, consumerId0, zkConnector); ZkCoordinator zkCoordinator1 = new StaticCoordinator("0,4,5", topics1, groupId1, consumerId1, zkConnector); Map<String, Integer> topicCount0 = zkCoordinator0.getRebalanceCount(); Map<String, Integer> topicCount1 = zkCoordinator1.getRebalanceCount(); assertTrue(topicCount0.get(topic0) == 3); assertTrue(topicCount0.get(topic1) == 3); assertTrue(topicCount1.get(topic0) == 2); assertTrue(topicCount1.get(topic1) == 2); zkCoordinator0.takePartition(topic0, 1); zkCoordinator1.takePartition(topic1, 0); zkCoordinator1.takePartition(topic1, 2); zkCoordinator1.takePartition(topic1, 4); topicCount0 = zkCoordinator0.getRebalanceCount(); topicCount1 = zkCoordinator1.getRebalanceCount(); assertTrue(topicCount0.get(topic0) == 2); assertTrue(topicCount0.get(topic1) == 3); assertTrue(topicCount1.get(topic0) == 2); assertTrue(topicCount1.get(topic1) == -1); }