@Test
public void testKMSAuthFailureRetry() throws Exception {

  Configuration conf = new Configuration();
  conf.set("hadoop.security.authentication", "kerberos");
  final File testDir = getTestDir();
  conf = createBaseKMSConf(testDir, conf);
  conf.set("hadoop.kms.authentication.kerberos.keytab",
      keytab.getAbsolutePath());
  conf.set("hadoop.kms.authentication.kerberos.principal", "HTTP/localhost");
  conf.set("hadoop.kms.authentication.kerberos.name.rules", "DEFAULT");
  conf.set("hadoop.kms.authentication.token.validity", "1");
  for (KMSACLs.Type type : KMSACLs.Type.values()) {
    conf.set(type.getAclConfigKey(), type.toString());
  }
  conf.set(KMSACLs.Type.CREATE.getAclConfigKey(),
      KMSACLs.Type.CREATE.toString() + ",SET_KEY_MATERIAL");
  conf.set(KMSACLs.Type.ROLLOVER.getAclConfigKey(),
      KMSACLs.Type.ROLLOVER.toString() + ",SET_KEY_MATERIAL");
  conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "k0.ALL", "*");
  conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "k1.ALL", "*");
  conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "k2.ALL", "*");
  conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "k3.ALL", "*");
  conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "k4.ALL", "*");
  writeConf(testDir, conf);
  runServer(null, null, testDir,
      new KMSCallable<Void>() {
        @Override
        public Void call() throws Exception {
          final Configuration conf = new Configuration();
          conf.setInt(KeyProvider.DEFAULT_BITLENGTH_NAME, 128);
          final URI uri = createKMSUri(getKMSUrl());
          doAs("SET_KEY_MATERIAL",
              new PrivilegedExceptionAction<Void>() {
                @Override
                public Void run() throws Exception {
                  KeyProvider kp = createProvider(uri, conf);
                  kp.createKey("k0", new byte[16],
                      new KeyProvider.Options(conf));
                  // This happens before rollover
                  kp.createKey("k1", new byte[16],
                      new KeyProvider.Options(conf));
                  // Atleast 2 rollovers.. so should induce signer Exception
                  Thread.sleep(3500);
                  kp.createKey("k2", new byte[16],
                    new KeyProvider.Options(conf));
                  return null;
                }
              });
          return null;
        }
      });
  // Test retry count
  runServer(null, null, testDir,
      new KMSCallable<Void>() {
        @Override
        public Void call() throws Exception {
          final Configuration conf = new Configuration();
          conf.setInt(KeyProvider.DEFAULT_BITLENGTH_NAME, 128);
          conf.setInt(KMSClientProvider.AUTH_RETRY, 0);
          final URI uri = createKMSUri(getKMSUrl());
          doAs("SET_KEY_MATERIAL",
              new PrivilegedExceptionAction<Void>() {
                @Override
                public Void run() throws Exception {
                  KeyProvider kp = createProvider(uri, conf);
                  kp.createKey("k3", new byte[16],
                      new KeyProvider.Options(conf));
                  // Atleast 2 rollovers.. so should induce signer Exception
                  Thread.sleep(3500);
                  try {
                    kp.createKey("k4", new byte[16],
                        new KeyProvider.Options(conf));
                    Assert.fail("This should not succeed..");
                  } catch (IOException e) {
                    Assert.assertTrue(
                        "HTTP exception must be a 401 : " + e.getMessage(), e
                            .getMessage().contains("401"));
                  }
                  return null;
                }
              });
          return null;
        }
      });
}