@Override
  public InetSocketAddress getPrimaryRpcAddress() throws UnavailableException {
    RetryPolicy retry = mRetryPolicySupplier.get();
    while (retry.attempt()) {
      InetSocketAddress address = getAddress();
      if (address != null) {
        return address;
      }
    }
    throw new UnavailableException(String.format(
        "Failed to determine primary master rpc address after polling each of %s %d times",
        mConnectDetails.getAddresses(), retry.getAttemptCount()));
  }
@Test(timeout = 10000)
  public void concurrentPollingMaster() throws Exception {
    int port1 = PortRegistry.reservePort();
    int port2 = PortRegistry.reservePort();
    InetSocketAddress serverAddress1 = new InetSocketAddress("127.0.0.1", port1);
    InetSocketAddress serverAddress2 = new InetSocketAddress("127.0.0.1", port2);
    RejectingServer s1 = new RejectingServer(serverAddress1, 20000);
    GrpcServer s2 =
        GrpcServerBuilder.forAddress(GrpcServerAddress.create(serverAddress2),
                new InstancedConfiguration(new AlluxioProperties()))
            .addService(ServiceType.META_MASTER_CLIENT_SERVICE, new GrpcService(
                new ServiceVersionClientServiceGrpc.ServiceVersionClientServiceImplBase() {
                })).build();
    try {
      s1.start();
      s2.start();
      List<InetSocketAddress> addrs =
          Arrays.asList(InetSocketAddress.createUnresolved("127.0.0.1", port1),
              InetSocketAddress.createUnresolved("127.0.0.1", port2));
      PollingMasterInquireClient client = new PollingMasterInquireClient(addrs,
          () -> new CountingRetry(0),
          new ConfigurationBuilder()
              .setProperty(PropertyKey.USER_MASTER_POLLING_CONCURRENT, true)
              .build(),
          ServiceType.META_MASTER_CLIENT_SERVICE);
      client.getPrimaryRpcAddress();
    } finally {
      s1.stopAndJoin();
      s2.shutdown();
      PortRegistry.release(port1);
      PortRegistry.release(port2);
    }
  }