@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)); }
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 { 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 { 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); }
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(); }
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();  }