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);  }
@Test public void readerShouldBeClosedAtTheEndOfExecution() throws Exception { job.call(); verify(reader).close(); }
@Test public void writerShouldBeClosedAtTheEndOfExecution() throws Exception { job.call(); verify(writer).close(); }
@Test public void whenNotAbleToOpenReader_ThenTheJobShouldFail() throws Exception { doThrow(exception).when(reader).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(); }
@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(); }
@Test public void whenNotAbleToOpenReader_thenTheJobListenerShouldBeInvoked() throws Exception { doThrow(exception).when(reader).open(); JobReport jobReportInternal = job.call(); InOrder inOrder = inOrder(jobListener1, jobListener2, reader, writer); inOrder.verify(jobListener2).afterJob(jobReportInternal); inOrder.verify(jobListener1).afterJob(jobReportInternal); inOrder.verify(reader).close(); inOrder.verify(writer).close(); }
@Test public void whenNotAbleToOpenWriter_thenTheJobListenerShouldBeInvoked() throws Exception { doThrow(exception).when(writer).open(); JobReport jobReport = job.call(); verify(jobListener1).afterJob(jobReport); verify(reader).close(); verify(writer).close(); }
@Test public void whenNotAbleToWriteRecords_ThenTheJobShouldFail() throws Exception { when(pipelineListener.beforeRecordProcessing(record1)).thenReturn(record1); when(pipelineListener.beforeRecordProcessing(record2)).thenReturn(record2); doThrow(exception).when(writer).writeRecords(new Batch(record1, record2)); JobReport jobReport = job.call(); assertThat(jobReport.getMetrics().getFilterCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getErrorCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getReadCount()).isEqualTo(2); assertThat(jobReport.getMetrics().getWriteCount()).isEqualTo(0); assertThat(jobReport.getStatus()).isEqualTo(JobStatus.FAILED); assertThat(jobReport.getLastError()).isEqualTo(exception); verify(reader).close(); verify(writer).close(); }
@Test public void reportShouldBeCorrect() { when(pipelineListener.beforeRecordProcessing(record1)).thenReturn(record1); when(pipelineListener.beforeRecordProcessing(record2)).thenReturn(record2); JobReport jobReport = job.call(); assertThat(jobReport.getMetrics().getFilterCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getErrorCount()).isEqualTo(0); assertThat(jobReport.getMetrics().getReadCount()).isEqualTo(2); assertThat(jobReport.getMetrics().getWriteCount()).isEqualTo(2); assertThat(jobReport.getStatus()).isEqualTo(JobStatus.COMPLETED); assertThat(jobReport.getLastError()).isNull(); }
@Test public void whenJobNameIsNotSpecified_thenTheJmxMBeanShouldBeRegisteredWithDefaultJobName() throws Exception { job = new JobBuilder<String, String>().enableJmx(true).build(); job.call(); MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); assertThat(mbs.isRegistered(new ObjectName(JobMonitorMBean.JMX_MBEAN_NAME + "name=" + JobParameters.DEFAULT_JOB_NAME))).isTrue(); }
@Test public void whenJobNameIsSpecified_thenTheJmxMBeanShouldBeRegisteredWithTheGivenJobName() throws Exception { String name = "master"; job = new JobBuilder<String, String>().enableJmx(true).named(name).build(); job.call(); MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); assertThat(mbs.isRegistered(new ObjectName(JobMonitorMBean.JMX_MBEAN_NAME + "name=" + name))).isTrue(); }
@Test public void jobListenerShouldBeInvoked() { job = new JobBuilder<String, String>() .reader(reader) .jobListener(jobListener1) .build(); JobReport report = job.call(); verify(jobListener1).beforeJob(any(JobParameters.class)); verify(jobListener1).afterJob(report); }
@Test public void whenRecordWriterThrowException_thenWriterListenerShouldBeInvoked() throws Exception { Batch<String> batch = new Batch<>(record1, record2); doThrow(exception).when(writer).writeRecords(batch); job = new JobBuilder<String, String>() .reader(reader) .writer(writer) .writerListener(recordWriterListener) .build(); job.call(); verify(recordWriterListener).onRecordWritingException(batch, exception); }
@Test public void pipelineListenerShouldBeInvokedForEachRecord() { when(pipelineListener.beforeRecordProcessing(record1)).thenReturn(record1); when(pipelineListener.beforeRecordProcessing(record2)).thenReturn(record2); job = new JobBuilder<String, String>() .reader(reader) .pipelineListener(pipelineListener) .build(); job.call(); verify(pipelineListener).beforeRecordProcessing(record1); verify(pipelineListener).afterRecordProcessing(record1, record1); verify(pipelineListener).beforeRecordProcessing(record2); verify(pipelineListener).afterRecordProcessing(record2, record2); }
@Test public void whenWriterThrowsExceptionAndBatchScanningIsActivated_thenShouldRewriteRecordsOneByOne() { class SavingRecordWriter implements RecordWriter<Integer> { private List<Batch<Integer>> batches = new ArrayList<>(); @Override public void writeRecords(Batch<Integer> batch) throws Exception { batches.add(batch); if (batch.size() == 2) { throw new Exception("Expected"); } } public List<Batch<Integer>> getBatches() { return batches; } } IterableRecordReader<Integer> recordReader = new IterableRecordReader<>(Arrays.asList(1, 2, 3, 4)); SavingRecordWriter recordWriter = new SavingRecordWriter(); Job job = new JobBuilder<Integer, Integer>() .batchSize(2) .enableBatchScanning(true) .reader(recordReader) .writer(recordWriter) .build(); job.call(); List<Batch<Integer>> batches = recordWriter.getBatches(); Assertions.assertThat(batches).hasSize(6); Batch<Integer> batch = batches.get(0); Iterator<Record<Integer>> iterator = batch.iterator(); Record<Integer> record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(1); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(1); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(2); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(2); Assertions.assertThat(iterator.hasNext()).isFalse(); batch = batches.get(1); iterator = batch.iterator(); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(1); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(1); Assertions.assertThat(iterator.hasNext()).isFalse(); batch = batches.get(2); iterator = batch.iterator(); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(2); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(2); Assertions.assertThat(iterator.hasNext()).isFalse(); batch = batches.get(3); iterator = batch.iterator(); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(3); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(3); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(4); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(4); Assertions.assertThat(iterator.hasNext()).isFalse(); batch = batches.get(4); iterator = batch.iterator(); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(3); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(3); Assertions.assertThat(iterator.hasNext()).isFalse(); batch = batches.get(5); iterator = batch.iterator(); record = iterator.next(); Assertions.assertThat(record.getHeader().getNumber()).isEqualTo(4); Assertions.assertThat(record.getHeader().isScanned()).isTrue(); Assertions.assertThat(record.getPayload()).isEqualTo(4); Assertions.assertThat(iterator.hasNext()).isFalse(); }