@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 FixedCountCoordinator(3, topics1, groupId1, consumerId0, zkConnector); ZkCoordinator zkCoordinator1 = new FixedCountCoordinator(1, topics1, groupId1, consumerId1, zkConnector); String path0 = zkCoordinator0.partitionPath(topic0); zkConnector.create(path0, false); zkConnector.create(path0 + "/0", false); zkConnector.create(path0 + "/1", false); zkConnector.create(path0 + "/2", false); zkConnector.create(path0 + "/3", false); String path1 = zkCoordinator0.partitionPath(topic1); zkConnector.create(path1, false); zkConnector.create(path1 + "/0", false); zkConnector.create(path1 + "/1", false); zkConnector.create(path1 + "/2", false); zkConnector.create(path1 + "/3", false); 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) == 1); assertTrue(topicCount1.get(topic1) == 1); zkCoordinator0.takePartition(topic0, 0); zkCoordinator1.takePartition(topic1, 0); zkCoordinator1.takePartition(topic1, 1); topicCount0 = zkCoordinator0.getRebalanceCount(); topicCount1 = zkCoordinator1.getRebalanceCount(); assertTrue(topicCount0.get(topic0) == 2); assertTrue(topicCount0.get(topic1) == 3); assertTrue(topicCount1.get(topic0) == 1); assertTrue(topicCount1.get(topic1) == -1); }
@Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> myPartitions = getMyPartitions(topic); countMap.put(topic, m_fixedCount - myPartitions.size()); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } }
FixedCountCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> myPartitions = getMyPartitions(topic); countMap.put(topic, m_fixedCount - myPartitions.size()); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } }
FixedCountCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> myPartitions = getMyPartitions(topic); countMap.put(topic, m_fixedCount - myPartitions.size()); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } FixedCountCoordinator(int fixedCount, List<String> topics,
			String groupId, String consumerId, KafkaController.ZkConnector zk); }
FixedCountCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> myPartitions = getMyPartitions(topic); countMap.put(topic, m_fixedCount - myPartitions.size()); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } FixedCountCoordinator(int fixedCount, 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); }
FixedCountCoordinator extends ZkCoordinator { @Override public synchronized Map<String, Integer> getRebalanceCount() { Map<String, Integer> countMap = new HashMap<String, Integer>(); try { for (String topic : m_allTopicPartitions.keySet()) { Set<Integer> myPartitions = getMyPartitions(topic); countMap.put(topic, m_fixedCount - myPartitions.size()); } return countMap; } catch (RuntimeException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } FixedCountCoordinator(int fixedCount, 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);  }