static Long replicationThrottle(CruiseControlRequestContext requestContext, KafkaCruiseControlConfig config) {
    Long value = getLongParam(requestContext, REPLICATION_THROTTLE_PARAM, config.getLong(ExecutorConfig.DEFAULT_REPLICATION_THROTTLE_CONFIG));
    if (value != null && value < 0) {
      throw new UserRequestException(String.format("Requested rebalance throttle must be non-negative (Requested: %s).", value));
    }
    return value;
  }
@Test
  public void testParseReplicationThrottleWithDefault() {
    CruiseControlRequestContext mockRequest = EasyMock.mock(CruiseControlRequestContext.class);
    KafkaCruiseControlConfig controlConfig = EasyMock.mock(KafkaCruiseControlConfig.class);
    // No parameter string value in the parameter map
    EasyMock.expect(mockRequest.getParameterMap()).andReturn(Collections.emptyMap()).once();
    EasyMock.expect(controlConfig.getLong(ExecutorConfig.DEFAULT_REPLICATION_THROTTLE_CONFIG))
            .andReturn(Long.valueOf(DEFAULT_REPLICATION_THROTTLE_STRING));

    EasyMock.replay(mockRequest, controlConfig);

    Long replicationThrottle = ParameterUtils.replicationThrottle(mockRequest, controlConfig);
    Assert.assertEquals(Long.valueOf(DEFAULT_REPLICATION_THROTTLE_STRING), replicationThrottle);
    EasyMock.verify(mockRequest, controlConfig);
  }