DynamicBrokersReader { public GlobalPartitionInformation getBrokerInfo() throws SocketTimeoutException { GlobalPartitionInformation globalPartitionInformation = new GlobalPartitionInformation(); try { int numPartitionsForTopic = getNumPartitions(); String brokerInfoPath = brokerPath(); for (int partition = 0; partition < numPartitionsForTopic; partition++) { int leader = getLeaderFor(partition); String path = brokerInfoPath + "/" + leader; try { byte[] brokerData = _curator.getData().forPath(path); Broker hp = getBrokerHost(brokerData); globalPartitionInformation.addPartition(partition, hp); } catch (org.apache.zookeeper.KeeperException.NoNodeException e) { LOG.error("Node {} does not exist ", path); } } } catch (SocketTimeoutException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } LOG.info("Read partition info from zookeeper: " + globalPartitionInformation); return globalPartitionInformation; } DynamicBrokersReader(Map conf, String zkStr, String zkPath, String topic); GlobalPartitionInformation getBrokerInfo(); String partitionPath(); String brokerPath(); void close(); static final Logger LOG; }
@Test public void testGetBrokerInfo() throws Exception { String host = "localhost"; int port = 9092; int partition = 0; addPartition(partition, host, port); GlobalPartitionInformation brokerInfo = dynamicBrokersReader.getBrokerInfo(); assertEquals(1, brokerInfo.getOrderedPartitions().size()); assertEquals(port, brokerInfo.getBrokerFor(partition).port); assertEquals(host, brokerInfo.getBrokerFor(partition).host); }
@Test public void testMultiplePartitionsOnDifferentHosts() throws Exception { String host = "localhost"; int port = 9092; int secondPort = 9093; int partition = 0; int secondPartition = partition + 1; addPartition(partition, 0, host, port); addPartition(secondPartition, 1, host, secondPort); GlobalPartitionInformation brokerInfo = dynamicBrokersReader.getBrokerInfo(); assertEquals(2, brokerInfo.getOrderedPartitions().size()); assertEquals(port, brokerInfo.getBrokerFor(partition).port); assertEquals(host, brokerInfo.getBrokerFor(partition).host); assertEquals(secondPort, brokerInfo.getBrokerFor(secondPartition).port); assertEquals(host, brokerInfo.getBrokerFor(secondPartition).host); }
@Test public void testMultiplePartitionsOnSameHost() throws Exception { String host = "localhost"; int port = 9092; int partition = 0; int secondPartition = partition + 1; addPartition(partition, 0, host, port); addPartition(secondPartition, 0, host, port); GlobalPartitionInformation brokerInfo = dynamicBrokersReader.getBrokerInfo(); assertEquals(2, brokerInfo.getOrderedPartitions().size()); assertEquals(port, brokerInfo.getBrokerFor(partition).port); assertEquals(host, brokerInfo.getBrokerFor(partition).host); assertEquals(port, brokerInfo.getBrokerFor(secondPartition).port); assertEquals(host, brokerInfo.getBrokerFor(secondPartition).host); }
@Test public void testSwitchHostForPartition() throws Exception { String host = "localhost"; int port = 9092; int partition = 0; addPartition(partition, host, port); GlobalPartitionInformation brokerInfo = dynamicBrokersReader.getBrokerInfo(); assertEquals(port, brokerInfo.getBrokerFor(partition).port); assertEquals(host, brokerInfo.getBrokerFor(partition).host); String newHost = host + "switch"; int newPort = port + 1; addPartition(partition, newHost, newPort); brokerInfo = dynamicBrokersReader.getBrokerInfo(); assertEquals(newPort, brokerInfo.getBrokerFor(partition).port); assertEquals(newHost, brokerInfo.getBrokerFor(partition).host); }