OperatorThread extends Thread { private void recreateWatchdogs() { List<OperationWatchdog> watchdogs = _context.getMasterQueue().getWatchdogs(); for (OperationWatchdog watchdog : watchdogs) { if (watchdog.isDone()) { LOG.info("release done watchdog " + watchdog); _queue.removeWatchdog(watchdog); } else { _registry.watchFor(watchdog); } } } OperatorThread(final MasterContext context, final long safeModeMaxTime); boolean isInSafeMode(); OperationRegistry getOperationRegistry(); MasterContext getContext(); @Override void run();  }
@Test(timeout = 10000) public void testRecreateWatchdogs() throws Exception { when(_queue.peek()).thenAnswer(new SleepingAnswer()); when(_protocol.getLiveNodes()).thenReturn(Arrays.asList("node1")); OperationWatchdog watchdog1 = mock(OperationWatchdog.class); OperationWatchdog watchdog2 = mock(OperationWatchdog.class); when(watchdog1.isDone()).thenReturn(true); when(watchdog2.isDone()).thenReturn(false); when(_queue.getWatchdogs()).thenReturn(Arrays.asList(watchdog1, watchdog2)); OperatorThread operatorThread = new OperatorThread(_context, 100); operatorThread.start(); waitUntilLeaveSafeMode(operatorThread); Thread.sleep(200); verify(_queue).getWatchdogs(); verify(watchdog1).isDone(); verify(watchdog2).isDone(); verify(_queue, times(1)).removeWatchdog(watchdog1); verify(_queue, times(0)).removeWatchdog(watchdog2); assertTrue(operatorThread.getOperationRegistry().getRunningOperations().contains(watchdog2.getOperation())); }