ReencryptionHandler implements Runnable { public ReencryptionPendingInodeIdCollector getTraverser() { return traverser; } ReencryptionHandler(final EncryptionZoneManager ezMgr,
      final Configuration conf); @Override void run(); ReencryptionPendingInodeIdCollector getTraverser(); static final Logger LOG; }
@Test public void testThrottle() throws Exception { final Configuration conf = new Configuration(); conf.setDouble(DFS_NAMENODE_REENCRYPT_THROTTLE_LIMIT_HANDLER_RATIO_KEY, 0.5); final ReencryptionHandler rh = mockReencryptionhandler(conf); final StopWatch mockAll = Mockito.mock(StopWatch.class); Mockito.when(mockAll.now(TimeUnit.MILLISECONDS)).thenReturn((long) 30000); Mockito.when(mockAll.reset()).thenReturn(mockAll); final StopWatch mockLocked = Mockito.mock(StopWatch.class); Mockito.when(mockLocked.now(TimeUnit.MILLISECONDS)) .thenReturn((long) 20000); Mockito.when(mockLocked.reset()).thenReturn(mockLocked); final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(); Whitebox.setInternalState(rh, "throttleTimerAll", mockAll); Whitebox.setInternalState(rh, "throttleTimerLocked", mockLocked); Whitebox.setInternalState(rh, "taskQueue", queue); final StopWatch sw = new StopWatch().start(); rh.getTraverser().throttle(); sw.stop(); assertTrue("should have throttled for at least 8 second", sw.now(TimeUnit.MILLISECONDS) > 8000); assertTrue("should have throttled for at most 12 second", sw.now(TimeUnit.MILLISECONDS) < 12000); }
@Test public void testThrottleNoOp() throws Exception { final Configuration conf = new Configuration(); conf.setDouble(DFS_NAMENODE_REENCRYPT_THROTTLE_LIMIT_HANDLER_RATIO_KEY, 0.5); final ReencryptionHandler rh = mockReencryptionhandler(conf); StopWatch mockAll = Mockito.mock(StopWatch.class); Mockito.when(mockAll.now()).thenReturn(new Long(30000)); Mockito.when(mockAll.reset()).thenReturn(mockAll); StopWatch mockLocked = Mockito.mock(StopWatch.class); Mockito.when(mockLocked.now()).thenReturn(new Long(10000)); Mockito.when(mockLocked.reset()).thenReturn(mockLocked); final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(); Whitebox.setInternalState(rh, "throttleTimerAll", mockAll); Whitebox.setInternalState(rh, "throttleTimerLocked", mockLocked); Whitebox.setInternalState(rh, "taskQueue", queue); final Map<Long, ReencryptionUpdater.ZoneSubmissionTracker> submissions = new HashMap<>(); Whitebox.setInternalState(rh, "submissions", submissions); StopWatch sw = new StopWatch().start(); rh.getTraverser().throttle(); sw.stop(); assertTrue("should not have throttled", sw.now(TimeUnit.MILLISECONDS) < 1000); }
@Test public void testThrottleAccumulatingTasks() throws Exception { final Configuration conf = new Configuration(); final ReencryptionHandler rh = mockReencryptionhandler(conf); 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); }