public static <T> T createRMProxy(final Configuration configuration,
      final Class<T> protocol) throws IOException {
    long rmConnectWait =
        configuration.getLong(
            YarnConfiguration.RESOURCEMANAGER_CONNECT_MAX_WAIT_MS,
            YarnConfiguration.DEFAULT_RESOURCEMANAGER_CONNECT_MAX_WAIT_MS);
    long rmRetryInterval =
        configuration.getLong(
            YarnConfiguration.RESOURCEMANAGER_CONNECT_RETRY_INTERVAL_MS,
            YarnConfiguration
                .DEFAULT_RESOURCEMANAGER_CONNECT_RETRY_INTERVAL_MS);
    long nmRmConnectWait =
        configuration.getLong(
            YarnConfiguration.NM_RESOURCEMANAGER_CONNECT_MAX_WAIT_MS,
                rmConnectWait);
    long nmRmRetryInterval =
        configuration.getLong(
            YarnConfiguration.NM_RESOURCEMANAGER_CONNECT_RETRY_INTERVAL_MS,
                rmRetryInterval);
    ServerRMProxy<T> serverRMProxy = new ServerRMProxy<>();
    return createRMProxy(configuration, protocol, serverRMProxy,
        nmRmConnectWait, nmRmRetryInterval);
  }
@Test
  public void testDistributedProtocol() {

    YarnConfiguration conf = new YarnConfiguration();
    try {
      ServerRMProxy.createRMProxy(conf, DistributedSchedulingAMProtocol.class);
    } catch (Exception e) {
      Assert.fail("DistributedSchedulingAMProtocol fail in non HA");
    }

    // HA is enabled
    conf.setBoolean(YarnConfiguration.RM_HA_ENABLED, true);
    conf.set(YarnConfiguration.RM_HA_IDS, "rm1,rm2");
    conf.set(HAUtil.addSuffix(YarnConfiguration.RM_HOSTNAME, "rm1"), "0.0.0.0");
    conf.set(HAUtil.addSuffix(YarnConfiguration.RM_HOSTNAME, "rm2"), "0.0.0.0");
    try {
      ServerRMProxy.createRMProxy(conf, DistributedSchedulingAMProtocol.class);
    } catch (Exception e) {
      Assert.fail("DistributedSchedulingAMProtocol fail in HA");
    }
  }