OperatorThread extends Thread { public boolean isInSafeMode() { return _safeMode; } OperatorThread(final MasterContext context, final long safeModeMaxTime); boolean isInSafeMode(); OperationRegistry getOperationRegistry(); MasterContext getContext(); @Override void run();  }
@Test(timeout = 10000) public void testSafeMode() throws Exception { final MasterOperation operation = mockOperation(ExecutionInstruction.EXECUTE); when(_queue.peek()).thenReturn(operation).thenAnswer(new SleepingAnswer()); when(_protocol.getLiveNodes()).thenReturn(EMPTY_LIST); long safeModeMaxTime = 200; OperatorThread operatorThread = new OperatorThread(_context, safeModeMaxTime); operatorThread.start(); Thread.sleep(safeModeMaxTime + 200); assertTrue(operatorThread.isAlive()); assertTrue(operatorThread.isInSafeMode()); when(_protocol.getLiveNodes()).thenReturn(Arrays.asList("node1")); Thread.sleep(safeModeMaxTime + 200); assertTrue(operatorThread.isAlive()); assertFalse(operatorThread.isInSafeMode()); verify(operation, times(1)).execute(_context, EMPTY_LIST); operatorThread.interrupt(); }
@Test(timeout = 10000) public void testGracefulShutdownWhileInSafeMode() throws Exception { when(_queue.peek()).thenAnswer(new SleepingAnswer()); long safeModeMaxTime = 2000; OperatorThread operatorThread = new OperatorThread(_context, safeModeMaxTime); operatorThread.start(); assertTrue(operatorThread.isAlive()); assertTrue(operatorThread.isInSafeMode()); operatorThread.interrupt(); operatorThread.join(); }
@Test(timeout = 10000) public void testGracefulShutdownWhileWaitingForOperations() throws Exception { when(_queue.peek()).thenAnswer(new SleepingAnswer()); when(_protocol.getLiveNodes()).thenReturn(Arrays.asList("node1")); long safeModeMaxTime = 200; OperatorThread operatorThread = new OperatorThread(_context, safeModeMaxTime); operatorThread.start(); waitUntilLeaveSafeMode(operatorThread); assertTrue(operatorThread.isAlive()); assertFalse(operatorThread.isInSafeMode()); operatorThread.interrupt(); operatorThread.join(); }