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