private synchronized void createOrUpdateDatabaseRateLimiter(List<String> databaseNames) {
    ExternalView brokerResource = HelixHelper
        .getExternalViewForResource(_helixManager.getClusterManagmentTool(), _helixManager.getClusterName(),
            CommonConstants.Helix.BROKER_RESOURCE_INSTANCE);
    for (String databaseName : databaseNames) {
      double databaseQpsQuota = getEffectiveQueryQuotaOnDatabase(databaseName);
      if (databaseQpsQuota < 0) {
        buildEmptyOrResetDatabaseRateLimiter(databaseName);
        continue;
      }
      int numOnlineBrokers = getNumOnlineBrokers(databaseName, brokerResource);
      double perBrokerQpsQuota = databaseQpsQuota / numOnlineBrokers;
      QueryQuotaEntity oldQueryQuotaEntity = _databaseRateLimiterMap.get(databaseName);
      if (oldQueryQuotaEntity == null) {
        LOGGER.info("Adding new query rate limiter for database {} with rate {}.", databaseName, perBrokerQpsQuota);
        QueryQuotaEntity queryQuotaEntity = new QueryQuotaEntity(RateLimiter.create(perBrokerQpsQuota),
            new HitCounter(ONE_SECOND_TIME_RANGE_IN_SECOND), new MaxHitRateTracker(ONE_MINUTE_TIME_RANGE_IN_SECOND),
            numOnlineBrokers, databaseQpsQuota, -1);
        _databaseRateLimiterMap.put(databaseName, queryQuotaEntity);
        continue;
      }
      boolean changeDetected = false;
      double oldQuota = oldQueryQuotaEntity.getRateLimiter() != null ? oldQueryQuotaEntity.getRateLimiter().getRate()
          : -1;
      if (oldQueryQuotaEntity.getOverallRate() != databaseQpsQuota) {
        changeDetected = true;
        LOGGER.info("Overall quota changed for the database from {} to {}", oldQueryQuotaEntity.getOverallRate(),
            databaseQpsQuota);
        oldQueryQuotaEntity.setOverallRate(databaseQpsQuota);
      }
      if (oldQueryQuotaEntity.getNumOnlineBrokers() != numOnlineBrokers) {
        changeDetected = true;
        LOGGER.info("Number of online brokers changed for the database from {} to {}",
            oldQueryQuotaEntity.getNumOnlineBrokers(), numOnlineBrokers);
        oldQueryQuotaEntity.setNumOnlineBrokers(numOnlineBrokers);
      }
      if (!changeDetected) {
        LOGGER.info("No change detected with the query rate limiter for database {}", databaseName);
        continue;
      }
      LOGGER.info("Updating existing query rate limiter for database {} from rate {} to {}", databaseName, oldQuota,
          perBrokerQpsQuota);
      oldQueryQuotaEntity.setRateLimiter(RateLimiter.create(perBrokerQpsQuota));
    }
  }
@Test
  public void testCreateOrUpdateDatabaseRateLimiter() {
    List<String> dbList = new ArrayList<>(2);
    dbList.add("db1");
    dbList.add("db2");
    dbList.add("db3");
    DatabaseConfig db1 = new DatabaseConfig(dbList.get(0), new QuotaConfig(null, null));
    DatabaseConfig db2 = new DatabaseConfig(dbList.get(1), new QuotaConfig(null, "1"));
    DatabaseConfig db3 = new DatabaseConfig(dbList.get(2), new QuotaConfig(null, "2"));
    ZKMetadataProvider.setDatabaseConfig(_testPropertyStore, db1);
    ZKMetadataProvider.setDatabaseConfig(_testPropertyStore, db2);
    ZKMetadataProvider.setDatabaseConfig(_testPropertyStore, db3);
    dbList.forEach(db -> _queryQuotaManager.createDatabaseRateLimiter(db));
    Map<String, QueryQuotaEntity> dbQuotaMap = _queryQuotaManager.getDatabaseRateLimiterMap();
    Assert.assertNull(dbQuotaMap.get(dbList.get(0)).getRateLimiter());
    Assert.assertEquals(dbQuotaMap.get(dbList.get(1)).getRateLimiter().getRate(), 1);
    Assert.assertEquals(dbQuotaMap.get(dbList.get(2)).getRateLimiter().getRate(), 2);
    db1.setQuotaConfig(new QuotaConfig(null, "1"));
    db2.setQuotaConfig(new QuotaConfig(null, "2"));
    ZKMetadataProvider.setDatabaseConfig(_testPropertyStore, db1);
    ZKMetadataProvider.setDatabaseConfig(_testPropertyStore, db2);
    dbList.forEach(db -> _queryQuotaManager.updateDatabaseRateLimiter(db));
    Assert.assertEquals(dbQuotaMap.get(dbList.get(0)).getRateLimiter().getRate(), 1);
    Assert.assertEquals(dbQuotaMap.get(dbList.get(1)).getRateLimiter().getRate(), 2);
    Assert.assertEquals(dbQuotaMap.get(dbList.get(2)).getRateLimiter().getRate(), 2);
  }