@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.");
    }
  }
@Test
  public void testParseConfigJbodFile() throws TimeoutException, BrokerCapacityResolutionException {
    BrokerCapacityConfigResolver configResolver = getBrokerCapacityConfigResolver("testCapacityConfigJBOD.json", this.getClass());

    assertEquals(2000000.0, configResolver.capacityForBroker("", "", 0, BROKER_CAPACITY_FETCH_TIMEOUT_MS, false)
                                         .capacity().get(Resource.DISK), 0.01);
    assertEquals(2200000.0, configResolver.capacityForBroker("", "", 3, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true)
                                         .capacity().get(Resource.DISK), 0.01);
    assertEquals(200000.0, configResolver.capacityForBroker("", "", 3, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true)
                                         .diskCapacityByLogDir().get("/tmp/kafka-logs-4"), 0.01);

    assertFalse(configResolver.capacityForBroker("", "", 2, BROKER_CAPACITY_FETCH_TIMEOUT_MS, false).isEstimated());
    assertTrue(configResolver.capacityForBroker("", "", 3, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true).isEstimated());
    assertTrue(configResolver.capacityForBroker("", "", 3, BROKER_CAPACITY_FETCH_TIMEOUT_MS, true).estimationInfo().length() > 0);
  }