public ConsumptionRateLimiter createRateLimiter(StreamConfig streamConfig, String tableName,
      ServerMetrics serverMetrics, String metricKeyName) {
    if (streamConfig.getTopicConsumptionRateLimit().isEmpty()) {
      return NOOP_RATE_LIMITER;
    }
    int partitionCount;
    try {
      partitionCount = _streamConfigToTopicPartitionCountMap.get(streamConfig);
    } catch (ExecutionException e) {
      // Exception here means for some reason, partition count cannot be fetched from stream!
      throw new RuntimeException(e);
    }
    double topicRateLimit = streamConfig.getTopicConsumptionRateLimit().get();
    double partitionRateLimit = topicRateLimit / partitionCount;
    LOGGER.info("A consumption rate limiter is set up for topic {} in table {} with rate limit: {} "
            + "(topic rate limit: {}, partition count: {})", streamConfig.getTopicName(), tableName, partitionRateLimit,
        topicRateLimit, partitionCount);
    MetricEmitter metricEmitter = new MetricEmitter(serverMetrics, metricKeyName);
    return new RateLimiterImpl(partitionRateLimit, metricEmitter);
  }
@Test
  public void testCreateRateLimiter() {
    // topic A
    ConsumptionRateLimiter rateLimiter = _consumptionRateManager.createRateLimiter(STREAM_CONFIG_A, TABLE_NAME);
    assertEquals(5.0, ((RateLimiterImpl) rateLimiter).getRate(), DELTA);

    // topic B
    rateLimiter = _consumptionRateManager.createRateLimiter(STREAM_CONFIG_B, TABLE_NAME);
    assertEquals(2.5, ((RateLimiterImpl) rateLimiter).getRate(), DELTA);

    // topic C
    rateLimiter = _consumptionRateManager.createRateLimiter(STREAM_CONFIG_C, TABLE_NAME);
    assertEquals(rateLimiter, NOOP_RATE_LIMITER);
  }