@Test
public void testThrottleAccumulatingTasks() throws Exception {

  final Configuration conf = new Configuration();
  final ReencryptionHandler rh = mockReencryptionhandler(conf);
  // mock tasks piling up
  final Map<Long, ReencryptionUpdater.ZoneSubmissionTracker>
      submissions = new HashMap<>();
  final ReencryptionUpdater.ZoneSubmissionTracker zst =
      new ReencryptionUpdater.ZoneSubmissionTracker();
  submissions.put(new Long(1), zst);
  Future mock = Mockito.mock(Future.class);
  for (int i = 0; i < Runtime.getRuntime().availableProcessors() * 3; ++i) {
    zst.addTask(mock);
  }
  Thread removeTaskThread = new Thread() {
    public void run() {
      try {
        Thread.sleep(3000);
      } catch (InterruptedException ie) {
        LOG.info("removeTaskThread interrupted.");
        Thread.currentThread().interrupt();
      }
      zst.getTasks().clear();
    }
  };
  Whitebox.setInternalState(rh, "submissions", submissions);
  final StopWatch sw = new StopWatch().start();
  removeTaskThread.start();
  rh.getTraverser().throttle();
  sw.stop();
  LOG.info("Throttle completed, consumed {}", sw.now(TimeUnit.MILLISECONDS));
  assertTrue("should have throttled for at least 3 second",
      sw.now(TimeUnit.MILLISECONDS) >= 3000);
}