DynamicCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { List<String> allConsumers = getAllConsumers(); m_thisConsumerIndex = getConsumerIndex(allConsumers); int consumerCount = allConsumers.size(); if (consumerCount <= 0) { LOGGER.error( "No consumers registered at node " + m_consumerPath); return countMap; } for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> oldAll = m_allTopicPartitions.get(topic); Set<Integer> newAll = getAllPartitions(topic); if (oldAll != null && newAll != null) { Set<Integer> newOnes = new HashSet<Integer>(newAll); newOnes.removeAll(oldAll); Set<Integer> newPartitions = m_newTopicPartitions.get(topic); if (newPartitions != null && !newOnes.isEmpty()) { newPartitions.addAll(newOnes); } } int partitionCount = newAll.size(); int minTake = partitionCount / consumerCount; int left = partitionCount % consumerCount; int shouldTake = minTake; if (m_thisConsumerIndex < left) shouldTake++; Set<Integer> myPartitions = getMyPartitions(topic); int takenCount = myPartitions.size(); int rebalanceCount = shouldTake - takenCount; countMap.put(topic, rebalanceCount); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } DynamicCoordinator(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"; ZkCoordinator zkCoordinator0 = new DynamicCoordinator(topics1, groupId1, consumerId0, zkConnector); ZkCoordinator zkCoordinator1 = new DynamicCoordinator(topics1, groupId1, consumerId1, zkConnector); try { zkCoordinator0.getRebalanceCount(); } catch (Exception e) { assertTrue(e instanceof RuntimeException); } zkCoordinator0.registerConsumer(); Map<String, Integer> topicCount0 = zkCoordinator0.getRebalanceCount(); assertTrue(topicCount0.get(topic0) == 0); assertTrue(topicCount0.get(topic1) == 0); String path0 = zkCoordinator0.partitionPath(topic0); zkConnector.create(path0, false); zkConnector.create(path0 + "/0", false); zkConnector.create(path0 + "/1", false); zkConnector.create(path0 + "/2", false); topicCount0 = zkCoordinator0.getRebalanceCount(); assertTrue(topicCount0.get(topic0) == 3); assertTrue(topicCount0.get(topic1) == 0); String path1 = zkCoordinator0.partitionPath(topic1); zkConnector.create(path1, false); zkConnector.create(path1 + "/0", false); zkConnector.create(path1 + "/1", false); zkCoordinator1.registerConsumer(); topicCount0 = zkCoordinator0.getRebalanceCount(); Map<String, Integer> topicCount1 = zkCoordinator1.getRebalanceCount(); int topic0TakenByC0 = topicCount0.get(topic0); int topic1TakenByC0 = topicCount0.get(topic1); int topic0TakenByC1 = topicCount1.get(topic0); int topic1TakenByC1 = topicCount1.get(topic1); assertTrue(topic0TakenByC0 + topic0TakenByC1 == 3); assertTrue(Math.abs(topic0TakenByC0 - topic0TakenByC1) <= 1); assertTrue(topic1TakenByC0 == 1); assertTrue(topic1TakenByC1 == 1); }