@Test
public void testClientRetriesSucceedsSecondTime() throws Exception {

  Configuration conf = new Configuration();
  conf.setInt(
      CommonConfigurationKeysPublic.KMS_CLIENT_FAILOVER_MAX_RETRIES_KEY, 3);
  KMSClientProvider p1 = mock(KMSClientProvider.class);
  when(p1.createKey(Mockito.anyString(), Mockito.any(Options.class)))
      .thenThrow(new ConnectTimeoutException("p1"))
      .thenReturn(new KMSClientProvider.KMSKeyVersion("test3", "v1",
              new byte[0]));
  KMSClientProvider p2 = mock(KMSClientProvider.class);
  when(p2.createKey(Mockito.anyString(), Mockito.any(Options.class)))
      .thenThrow(new ConnectTimeoutException("p2"));
  when(p1.getKMSUrl()).thenReturn("p1");
  when(p2.getKMSUrl()).thenReturn("p2");
  LoadBalancingKMSClientProvider kp = new LoadBalancingKMSClientProvider(
      new KMSClientProvider[] {p1, p2}, 0, conf);
  try {
    kp.createKey("test3", new Options(conf));
  } catch (Exception e) {
    fail("Provider p1 should have answered the request second time.");
  }
  verify(p1, Mockito.times(2)).createKey(Mockito.eq("test3"),
          Mockito.any(Options.class));
  verify(p2, Mockito.times(1)).createKey(Mockito.eq("test3"),
          Mockito.any(Options.class));
}