@SuppressWarnings("unchecked")
@Test
public void testNodeUpAferAWhile() throws Exception {

  for (int i = 0; i < TEST_RETRIES; i++) {
    String connectString = zkServer.getConnectString();
    Configuration conf = getSecretConf(connectString);
    DelegationTokenManager tm1 = new DelegationTokenManager(conf, new Text("bla"));
    tm1.init();
    Token<DelegationTokenIdentifier> token1 =
        (Token<DelegationTokenIdentifier>) tm1.createToken(
            UserGroupInformation.getCurrentUser(), "foo");
    Assert.assertNotNull(token1);
    Token<DelegationTokenIdentifier> token2 =
        (Token<DelegationTokenIdentifier>) tm1.createToken(
            UserGroupInformation.getCurrentUser(), "bar");
    Assert.assertNotNull(token2);
    Token<DelegationTokenIdentifier> token3 =
        (Token<DelegationTokenIdentifier>) tm1.createToken(
            UserGroupInformation.getCurrentUser(), "boo");
    Assert.assertNotNull(token3);
    tm1.verifyToken(token1);
    tm1.verifyToken(token2);
    tm1.verifyToken(token3);
    // Cancel one token
    tm1.cancelToken(token1, "foo");
    // Start second node after some time..
    Thread.sleep(1000);
    DelegationTokenManager tm2 = new DelegationTokenManager(conf, new Text("bla"));
    tm2.init();
    tm2.verifyToken(token2);
    tm2.verifyToken(token3);
    try {
      verifyTokenFail(tm2, token1);
      fail("Expected InvalidToken");
    } catch (SecretManager.InvalidToken it) {
      // Ignore
    }
    // Create a new token thru the new ZKDTSM
    Token<DelegationTokenIdentifier> token4 =
        (Token<DelegationTokenIdentifier>) tm2.createToken(
            UserGroupInformation.getCurrentUser(), "xyz");
    Assert.assertNotNull(token4);
    tm2.verifyToken(token4);
    tm1.verifyToken(token4);
    // Bring down tm2
    verifyDestroy(tm2, conf);
    // Start third node after some time..
    Thread.sleep(1000);
    DelegationTokenManager tm3 = new DelegationTokenManager(conf, new Text("bla"));
    tm3.init();
    tm3.verifyToken(token2);
    tm3.verifyToken(token3);
    tm3.verifyToken(token4);
    try {
      verifyTokenFail(tm3, token1);
      fail("Expected InvalidToken");
    } catch (SecretManager.InvalidToken it) {
      // Ignore
    }
    verifyDestroy(tm3, conf);
    verifyDestroy(tm1, conf);
  }
}