ZKDelegationTokenSecretManager extends AbstractDelegationTokenSecretManager<TokenIdent> { @Override public TokenIdent cancelToken(Token<TokenIdent> token, String canceller) throws IOException { ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier()); DataInputStream in = new DataInputStream(buf); TokenIdent id = createIdentifier(); id.readFields(in); syncLocalCacheWithZk(id); return super.cancelToken(token, canceller); } ZKDelegationTokenSecretManager(Configuration conf); static void setCurator(CuratorFramework curator); @Override void startThreads(); @Override void stopThreads(); @Override TokenIdent cancelToken(Token<TokenIdent> token,
      String canceller); boolean isTokenWatcherEnabled(); @VisibleForTesting ExecutorService getListenerThreadPool(); static final String ZK_CONF_PREFIX; static final String ZK_DTSM_ZK_NUM_RETRIES; static final String ZK_DTSM_ZK_SESSION_TIMEOUT; static final String ZK_DTSM_ZK_CONNECTION_TIMEOUT; static final String ZK_DTSM_ZK_SHUTDOWN_TIMEOUT; static final String ZK_DTSM_ZNODE_WORKING_PATH; static final String ZK_DTSM_ZK_AUTH_TYPE; static final String ZK_DTSM_ZK_CONNECTION_STRING; static final String ZK_DTSM_ZK_KERBEROS_KEYTAB; static final String ZK_DTSM_ZK_KERBEROS_PRINCIPAL; static final String ZK_DTSM_TOKEN_SEQNUM_BATCH_SIZE; static final String ZK_DTSM_TOKEN_WATCHER_ENABLED; static final boolean ZK_DTSM_TOKEN_WATCHER_ENABLED_DEFAULT; static final int ZK_DTSM_ZK_NUM_RETRIES_DEFAULT; static final int ZK_DTSM_ZK_SESSION_TIMEOUT_DEFAULT; static final int ZK_DTSM_ZK_CONNECTION_TIMEOUT_DEFAULT; static final int ZK_DTSM_ZK_SHUTDOWN_TIMEOUT_DEFAULT; static final String ZK_DTSM_ZNODE_WORKING_PATH_DEAFULT; static final int ZK_DTSM_TOKEN_SEQNUM_BATCH_SIZE_DEFAULT; }
@SuppressWarnings("unchecked") @Test public void testMultiNodeOperations() throws Exception { for (int i = 0; i < TEST_RETRIES; i++) { DelegationTokenManager tm1, tm2 = null; String connectString = zkServer.getConnectString(); Configuration conf = getSecretConf(connectString); tm1 = new DelegationTokenManager(conf, new Text("bla")); tm1.init(); tm2 = new DelegationTokenManager(conf, new Text("bla")); tm2.init(); Token<DelegationTokenIdentifier> token = (Token<DelegationTokenIdentifier>) tm1.createToken( UserGroupInformation.getCurrentUser(), "foo"); Assert.assertNotNull(token); tm2.verifyToken(token); tm2.renewToken(token, "foo"); tm1.verifyToken(token); tm1.cancelToken(token, "foo"); try { verifyTokenFail(tm2, token); fail("Expected InvalidToken"); } catch (SecretManager.InvalidToken it) { } token = (Token<DelegationTokenIdentifier>) tm2.createToken( UserGroupInformation.getCurrentUser(), "bar"); Assert.assertNotNull(token); tm1.verifyToken(token); tm1.renewToken(token, "bar"); tm2.verifyToken(token); tm2.cancelToken(token, "bar"); try { verifyTokenFail(tm1, token); fail("Expected InvalidToken"); } catch (SecretManager.InvalidToken it) { } verifyDestroy(tm1, conf); verifyDestroy(tm2, conf); } }
@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); tm1.cancelToken(token1, "foo"); 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) { } Token<DelegationTokenIdentifier> token4 = (Token<DelegationTokenIdentifier>) tm2.createToken( UserGroupInformation.getCurrentUser(), "xyz"); Assert.assertNotNull(token4); tm2.verifyToken(token4); tm1.verifyToken(token4); verifyDestroy(tm2, conf); 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) { } verifyDestroy(tm3, conf); verifyDestroy(tm1, conf); } }
@SuppressWarnings("unchecked") @Test public void testCancelTokenSingleManager() throws Exception { for (int i = 0; i < TEST_RETRIES; i++) { DelegationTokenManager tm1 = null; String connectString = zkServer.getConnectString(); Configuration conf = getSecretConf(connectString); tm1 = new DelegationTokenManager(conf, new Text("foo")); tm1.init(); Token<DelegationTokenIdentifier> token = (Token<DelegationTokenIdentifier>) tm1.createToken(UserGroupInformation.getCurrentUser(), "foo"); Assert.assertNotNull(token); tm1.cancelToken(token, "foo"); try { verifyTokenFail(tm1, token); fail("Expected InvalidToken"); } catch (SecretManager.InvalidToken it) { it.printStackTrace(); } verifyDestroy(tm1, conf); } }