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 testParseReplicationThrottleWithNoDefault() {
    CruiseControlRequestContext mockRequest = EasyMock.mock(CruiseControlRequestContext.class);
    KafkaCruiseControlConfig controlConfig = EasyMock.mock(KafkaCruiseControlConfig.class);

    Map<String, String[]> paramMap = Collections.singletonMap(
        ParameterUtils.REPLICATION_THROTTLE_PARAM,
        new String[]{ParameterUtils.REPLICATION_THROTTLE_PARAM});

    EasyMock.expect(mockRequest.getParameterMap()).andReturn(paramMap).once();
    EasyMock.expect(mockRequest.getParameter(ParameterUtils.REPLICATION_THROTTLE_PARAM)).andReturn(REPLICATION_THROTTLE_STRING).once();
    // No default
    EasyMock.expect(controlConfig.getLong(ExecutorConfig.DEFAULT_REPLICATION_THROTTLE_CONFIG)).andReturn(null);

    EasyMock.replay(mockRequest, controlConfig);

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