ZKDelegationTokenSecretManager extends AbstractDelegationTokenSecretManager<TokenIdent> { @Override public void stopThreads() { super.stopThreads(); try { if (tokenCache != null) { tokenCache.close(); } } catch (Exception e) { LOG.error("Could not stop Delegation Token Cache", e); } try { if (delTokSeqCounter != null) { delTokSeqCounter.close(); } } catch (Exception e) { LOG.error("Could not stop Delegation Token Counter", e); } try { if (keyIdSeqCounter != null) { keyIdSeqCounter.close(); } } catch (Exception e) { LOG.error("Could not stop Key Id Counter", e); } try { if (keyCache != null) { keyCache.close(); } } catch (Exception e) { LOG.error("Could not stop KeyCache", e); } try { if (!isExternalClient && (zkClient != null)) { zkClient.close(); } } catch (Exception e) { LOG.error("Could not stop Curator Framework", e); } if (listenerThreadPool != null) { listenerThreadPool.shutdown(); try { if (!listenerThreadPool.awaitTermination(shutdownTimeout, TimeUnit.MILLISECONDS)) { LOG.error("Forcing Listener threadPool to shutdown !!"); listenerThreadPool.shutdownNow(); } } catch (InterruptedException ie) { listenerThreadPool.shutdownNow(); Thread.currentThread().interrupt(); } } } ZKDelegationTokenSecretManager(Configuration conf); static void setCurator(CuratorFramework curator); @Override void startThreads(); @Override void stopThreads(); @Override synchronized TokenIdent cancelToken(Token<TokenIdent> token,
      String canceller); @VisibleForTesting ExecutorService getListenerThreadPool(); 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 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; }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testStopThreads() throws Exception { DelegationTokenManager tm1 = null; String connectString = zkServer.getConnectString(); final long updateIntervalSeconds = 1; final long shutdownTimeoutMillis = updateIntervalSeconds * 1000 * 5; Configuration conf = getSecretConf(connectString); conf.setLong(DelegationTokenManager.UPDATE_INTERVAL, updateIntervalSeconds); conf.setLong(DelegationTokenManager.REMOVAL_SCAN_INTERVAL, updateIntervalSeconds); conf.setLong(DelegationTokenManager.RENEW_INTERVAL, updateIntervalSeconds); conf.setLong(ZKDelegationTokenSecretManager.ZK_DTSM_ZK_SHUTDOWN_TIMEOUT, shutdownTimeoutMillis); tm1 = new DelegationTokenManager(conf, new Text("foo")); tm1.init(); Token<DelegationTokenIdentifier> token = (Token<DelegationTokenIdentifier>) tm1.createToken(UserGroupInformation.getCurrentUser(), "foo"); Assert.assertNotNull(token); AbstractDelegationTokenSecretManager sm = tm1.getDelegationTokenSecretManager(); ZKDelegationTokenSecretManager zksm = (ZKDelegationTokenSecretManager)sm; ExecutorService es = zksm.getListenerThreadPool(); es.submit(new Callable<Void>() { public Void call() throws Exception { Thread.sleep(shutdownTimeoutMillis * 2); return null; } }); tm1.destroy(); }