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