ProcessStageRunner { protected void performProcessing(LocalDocument doc) throws Exception { logger.debug("Got new doc '{}' to process.", doc.getID()); try { logger.trace("Waiting for processing of doc '{}'", doc.getID()); processWithTimeout(doc); logger.trace("Processing finished of doc '{}'", doc.getID()); } catch (ExecutionException e) { onException(doc, unwrapExecutionException(e)); return; } catch (TimeoutException e) { onException(doc, e); throw e; } catch (InterruptedException e) { logger.info("Processing was interrupted"); Thread.currentThread().interrupt(); return; } onSuccess(doc); } ProcessStageRunner(String stageName, AbstractProcessStage stage, RemotePipeline remotePipeline); void shutdownProcessing();  }
@Test public void testPerformProcessing_fails_document_if_stage_throws_exceptions() throws Exception { ProcessStageRunner stageRunner = buildStageRunner(new AbstractProcessStage() { @Override public void process(LocalDocument doc) throws Exception { throw new Exception("Stage failed for some reason"); } }); spy(stageRunner); LocalDocument testDoc1 = mock(LocalDocument.class); stageRunner.performProcessing(testDoc1); verify(rp, times(1)).markFailed(eq(testDoc1), any(Exception.class)); }
@Test public void testPerformProcessing_persists_error_on_save_failure() throws Exception { when(rp.save(any(LocalDocument.class))).thenReturn(false); when(rp.markFailed(any(LocalDocument.class), any(Throwable.class))).thenReturn(false); ProcessStageRunner stageRunner = buildStageRunner(new AbstractProcessStage() { @Override public void process(LocalDocument doc) {} }); stageRunner.performProcessing(new LocalDocument()); verify(rp, times(1)).save(any(LocalDocument.class)); verify(rp, times(1)).markFailed(any(LocalDocument.class), any(Throwable.class)); }
@Test(timeout = 1000) public void testPerformProcessing_throws_timeout_exception_if_timeout_exceeded() throws Exception { exception.expect(TimeoutException.class); HangingStage processStage = new HangingStage(); processStage.setProcessingTimeout(1); ProcessStageRunner stageRunner = buildStageRunner(processStage); stageRunner.performProcessing(mock(LocalDocument.class)); }
@Test public void testPerformProcessing_does_not_rethrow_timeout_exception_if_thrown_by_stage() throws Exception { ProcessStageRunner stageRunner = buildStageRunner(new AbstractProcessStage() { @Override public void process(LocalDocument doc) throws Exception { throw new TimeoutException(); } }); stageRunner.performProcessing(mock(LocalDocument.class)); }