@Test public void whenNotAbleToOpenWriter_ThenTheJobShouldFail() throws Exception { doThrow(exception).when(writer).open(); JobReport jobReport = job.call(); assertThat(jobReport).isNotNull(); assertThat(jobReport.getStatus()).isEqualTo(JobStatus.FAILED); assertThat(jobReport.getMetrics().getFilterCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getErrorCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getReadCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getWriteCount()).isEqualTo(0); assertThat(jobReport.getLastError()).isEqualTo(exception); verify(reader).close(); verify(writer).close(); }
@Override public JobReport call() { start(); try { openReader(); openWriter(); setStatus(JobStatus.STARTED); while (moreRecords() && !isInterrupted()) { Batch<O> batch = readAndProcessBatch(); writeBatch(batch); } setStatus(JobStatus.STOPPING); } catch (Exception exception) { fail(exception); return report; } finally { closeReader(); closeWriter(); } teardown(); return report; }
BatchJob implements Job { @Override public JobReport call() { start(); try { openReader(); openWriter(); setStatus(JobStatus.STARTED); while (moreRecords() && !isInterrupted()) { Batch<O> batch = readAndProcessBatch(); writeBatch(batch); } setStatus(JobStatus.STOPPING); } catch (Exception exception) { fail(exception); return report; } finally { closeReader(); closeWriter(); } teardown(); return report; } }
BatchJob implements Job { @Override public JobReport call() { start(); try { openReader(); openWriter(); setStatus(JobStatus.STARTED); while (moreRecords() && !isInterrupted()) { Batch<O> batch = readAndProcessBatch(); writeBatch(batch); } setStatus(JobStatus.STOPPING); } catch (Exception exception) { fail(exception); return report; } finally { closeReader(); closeWriter(); } teardown(); return report; } BatchJob(JobParameters parameters); }
BatchJob implements Job { @Override public JobReport call() { start(); try { openReader(); openWriter(); setStatus(JobStatus.STARTED); while (moreRecords() && !isInterrupted()) { Batch<O> batch = readAndProcessBatch(); writeBatch(batch); } setStatus(JobStatus.STOPPING); } catch (Exception exception) { fail(exception); return report; } finally { closeReader(); closeWriter(); } teardown(); return report; } BatchJob(JobParameters parameters); @Override String getName(); @Override JobReport call(); void setRecordReader(RecordReader<I> recordReader); void setRecordWriter(RecordWriter<O> recordWriter); void addRecordProcessor(RecordProcessor<?, ?> recordProcessor); void addBatchListener(BatchListener<O> batchListener); void addJobListener(JobListener jobListener); void addRecordReaderListener(RecordReaderListener<I> recordReaderListener); void addRecordWriterListener(RecordWriterListener<O> recordWriterListener); void addPipelineListener(PipelineListener pipelineListener); void setName(String name); }
BatchJob implements Job { @Override public JobReport call() { start(); try { openReader(); openWriter(); setStatus(JobStatus.STARTED); while (moreRecords() && !isInterrupted()) { Batch<O> batch = readAndProcessBatch(); writeBatch(batch); } setStatus(JobStatus.STOPPING); } catch (Exception exception) { fail(exception); return report; } finally { closeReader(); closeWriter(); } teardown(); return report; } BatchJob(JobParameters parameters); @Override String getName(); @Override JobReport call(); void setRecordReader(RecordReader<I> recordReader); void setRecordWriter(RecordWriter<O> recordWriter); void addRecordProcessor(RecordProcessor<?, ?> recordProcessor); void addBatchListener(BatchListener<O> batchListener); void addJobListener(JobListener jobListener); void addRecordReaderListener(RecordReaderListener<I> recordReaderListener); void addRecordWriterListener(RecordWriterListener<O> recordWriterListener); void addPipelineListener(PipelineListener pipelineListener); void setName(String name);  }