@Test(expected = BrokerCapacityResolutionException.class) public void testParseConfigFile() throws TimeoutException, BrokerCapacityResolutionException { BrokerCapacityConfigResolver configResolver = getBrokerCapacityConfigResolver("testCapacityConfig.json", this.getClass()); assertEquals(200000.0, configResolver.capacityForBroker("", "", 0, BROKER_CAPACITY_FETCH_TIMEOUT_MS, false) .capacity().get(Resource.NW_IN), 0.01); assertEquals(100000.0, configResolver.capacityForBroker("", "", 2, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true) .capacity().get(Resource.NW_IN), 0.01); try { configResolver.capacityForBroker("", "", BrokerCapacityConfigFileResolver.DEFAULT_CAPACITY_BROKER_ID, BROKER_CAPACITY_FETCH_TIMEOUT_MS, false); fail("Should have thrown exception for negative broker id"); } catch (IllegalArgumentException e) { } assertTrue(configResolver.capacityForBroker("", "", 2, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true).isEstimated()); assertTrue(configResolver.capacityForBroker("", "", 2, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true).estimationInfo().length() > 0); configResolver.capacityForBroker("", "", 2, BROKER_CAPACITY_FETCH_TIMEOUT_MS, false); }
@Override public BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation) throws BrokerCapacityResolutionException { if (brokerId >= 0) { BrokerCapacityInfo capacity = _capacitiesForBrokers.get(brokerId); if (capacity != null) { return capacity; } else { if (allowCapacityEstimation) { String info = String.format("Missing broker id(%d) in capacity config file.", brokerId); return new BrokerCapacityInfo(_capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).capacity(), info, _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).diskCapacityByLogDir(), _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).numCpuCores()); } else { throw new BrokerCapacityResolutionException(String.format("Unable to resolve capacity of broker %d. Either (1) adding the " + "default broker capacity (via adding capacity for broker %d and allowing capacity estimation) or (2) adding missing " + "broker's capacity in file %s.", brokerId, DEFAULT_CAPACITY_BROKER_ID, _configFile)); } } } else { throw new IllegalArgumentException("The broker id(" + brokerId + ") should be non-negative."); } }
BrokerCapacityConfigFileResolver implements BrokerCapacityConfigResolver { @Override public BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation) throws BrokerCapacityResolutionException { if (brokerId >= 0) { BrokerCapacityInfo capacity = _capacitiesForBrokers.get(brokerId); if (capacity != null) { return capacity; } else { if (allowCapacityEstimation) { String info = String.format("Missing broker id(%d) in capacity config file.", brokerId); return new BrokerCapacityInfo(_capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).capacity(), info, _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).diskCapacityByLogDir(), _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).numCpuCores()); } else { throw new BrokerCapacityResolutionException(String.format("Unable to resolve capacity of broker %d. Either (1) adding the " + "default broker capacity (via adding capacity for broker %d and allowing capacity estimation) or (2) adding missing " + "broker's capacity in file %s.", brokerId, DEFAULT_CAPACITY_BROKER_ID, _configFile)); } } } else { throw new IllegalArgumentException("The broker id(" + brokerId + ") should be non-negative."); } } }
BrokerCapacityConfigFileResolver implements BrokerCapacityConfigResolver { @Override public BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation) throws BrokerCapacityResolutionException { if (brokerId >= 0) { BrokerCapacityInfo capacity = _capacitiesForBrokers.get(brokerId); if (capacity != null) { return capacity; } else { if (allowCapacityEstimation) { String info = String.format("Missing broker id(%d) in capacity config file.", brokerId); return new BrokerCapacityInfo(_capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).capacity(), info, _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).diskCapacityByLogDir(), _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).numCpuCores()); } else { throw new BrokerCapacityResolutionException(String.format("Unable to resolve capacity of broker %d. Either (1) adding the " + "default broker capacity (via adding capacity for broker %d and allowing capacity estimation) or (2) adding missing " + "broker's capacity in file %s.", brokerId, DEFAULT_CAPACITY_BROKER_ID, _configFile)); } } } else { throw new IllegalArgumentException("The broker id(" + brokerId + ") should be non-negative."); } }  }
BrokerCapacityConfigFileResolver implements BrokerCapacityConfigResolver { @Override public BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation) throws BrokerCapacityResolutionException { if (brokerId >= 0) { BrokerCapacityInfo capacity = _capacitiesForBrokers.get(brokerId); if (capacity != null) { return capacity; } else { if (allowCapacityEstimation) { String info = String.format("Missing broker id(%d) in capacity config file.", brokerId); return new BrokerCapacityInfo(_capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).capacity(), info, _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).diskCapacityByLogDir(), _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).numCpuCores()); } else { throw new BrokerCapacityResolutionException(String.format("Unable to resolve capacity of broker %d. Either (1) adding the " + "default broker capacity (via adding capacity for broker %d and allowing capacity estimation) or (2) adding missing " + "broker's capacity in file %s.", brokerId, DEFAULT_CAPACITY_BROKER_ID, _configFile)); } } } else { throw new IllegalArgumentException("The broker id(" + brokerId + ") should be non-negative."); } }  @Override void configure(Map<String, ?> configs); @Override BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation); @Override void close(); }
BrokerCapacityConfigFileResolver implements BrokerCapacityConfigResolver { @Override public BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation) throws BrokerCapacityResolutionException { if (brokerId >= 0) { BrokerCapacityInfo capacity = _capacitiesForBrokers.get(brokerId); if (capacity != null) { return capacity; } else { if (allowCapacityEstimation) { String info = String.format("Missing broker id(%d) in capacity config file.", brokerId); return new BrokerCapacityInfo(_capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).capacity(), info, _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).diskCapacityByLogDir(), _capacitiesForBrokers.get(DEFAULT_CAPACITY_BROKER_ID).numCpuCores()); } else { throw new BrokerCapacityResolutionException(String.format("Unable to resolve capacity of broker %d. Either (1) adding the " + "default broker capacity (via adding capacity for broker %d and allowing capacity estimation) or (2) adding missing " + "broker's capacity in file %s.", brokerId, DEFAULT_CAPACITY_BROKER_ID, _configFile)); } } } else { throw new IllegalArgumentException("The broker id(" + brokerId + ") should be non-negative."); } }  @Override void configure(Map<String, ?> configs); @Override BrokerCapacityInfo capacityForBroker(String rack, String host, int brokerId, long timeoutMs, boolean allowCapacityEstimation); @Override void close(); static final String CAPACITY_CONFIG_FILE; static final int DEFAULT_CAPACITY_BROKER_ID; static final double DEFAULT_CPU_CAPACITY_WITH_CORES; }