@SuppressWarnings({ "unchecked", "rawtypes" })

@Test
public void testStopThreads() throws Exception {
  DelegationTokenManager tm1 = null;
  String connectString = zkServer.getConnectString();
  // let's make the update interval short and the shutdown interval
  // comparatively longer, so if the update thread runs after shutdown,
  // it will cause an error.
  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); // force this to be shutdownNow
      return null;
    }
  });
  tm1.destroy();
}