Spring Batch Using CompositeItemWriter and CompositeItemProcessor

Question

Using Spring Batch, I have to write in two different table, but using the same ItemReader.

I can't figure out how to use one ItemReader and a CompositeItemWriter.

Here's the JobConfiguration :

public class JobConfiguration {

    @Autowired
    @SuppressWarnings("squid:S3305")
    private ItemReaderSurveillance itemReaderSurveillance;

    @Autowired
    @SuppressWarnings("squid:S3305")
    private ItemWriterSurveillance itemWriterSurveillance;

    @Autowired
    @SuppressWarnings("squid:S3305")
    private ItemWriterSurveillanceEcheance itemWriterSurveillanceEcheance;

    @Autowired
    @SuppressWarnings("squid:S3305")
    private CompositeItemProcessorSurveillance compositeItemProcessor;

    @Bean(name = "importSurveillanceJob")
    public Job job(JobBuilderFactory jobs) {

        return jobs.get("importSurveillanceStep")
                .listener(jobListener)
                .start(stepTaskletCreationRepertoireReport())
                .next(stepTaskletCreationRepertoireArchive())
                .next(stepSurveillanceReadProcessWrite())
                .next(stepZipFile())
                .build();
    }

    @Bean
    protected Step stepSurveillanceReadProcessWrite() {
        return stepBuilderFactory.get("stepSurveillanceReadProcessWrite")
                .<SurveillanceLineFile, CompositeResultSurveillance>chunk(Integer.valueOf(commitInterval))
                .reader(itemReaderSurveillance)
                .processor(compositeItemProcessor)
                .writer(compositeItemWriter())
                .faultTolerant()
                .retryLimit(0)
                .build();
    }

    @Bean
    public CompositeItemWriter<CompositeResultSurveillance> compositeItemWriter(){
        CompositeItemWriter compositeItemWriter = new CompositeItemWriter();
        compositeItemWriter.setDelegates(Arrays.asList(itemWriterSurveillance, itemWriterSurveillanceEcheance));
        return compositeItemWriter;
    }
}

Item Writers :

@Slf4j
@StepScope
@Component
public class ItemWriterSurveillance implements ItemWriter<FoaSurveillance>, StepExecutionListener {
String fileName;
    JobExecution mJobExecution;
    StepExecution stepExecution;

    @Autowired
    private FoaSurveillanceDao foaSurveillanceDao;

    @Override
    public void write(List<? extends FoaSurveillance> foaSurveillances) {

        ExecutionContext executionContext = stepExecution.getExecutionContext();
        
        // Process data
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        mJobExecution = stepExecution.getJobExecution();
        this.stepExecution = stepExecution;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }
}

@Slf4j
@StepScope
@Component
public class ItemWriterSurveillanceEcheance implements ItemWriter<FoaSurveillanceEcheance>, StepExecutionListener {

    String fileName;
    JobExecution mJobExecution;
    StepExecution stepExecution;

    @Autowired
    private FoaSurveillanceEcheanceDao foaSurveillanceEcheanceDao;

    @Override
    public void write(List<? extends FoaSurveillanceEcheance> foaSurveillanceEcheances) {

        ExecutionContext executionContext = stepExecution.getExecutionContext();
        // Process data
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        mJobExecution = stepExecution.getJobExecution();
        this.stepExecution = stepExecution;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }
}

Compose the two itemProcessor :

@Slf4j
@Component
public class CompositeItemProcessorSurveillance implements ItemProcessor<SurveillanceLineFile, CompositeResultSurveillance>, StepExecutionListener {

    private StepExecution stepExecution;

    @Autowired
    ItemProcessorSurveillance itemProcessorSurveillance;

    @Autowired
    ItemProcessorSurveillanceEcheance itemProcessorSurveillanceEcheance;

    @Override
    public CompositeResultSurveillance process(SurveillanceLineFile surveillanceLineFile) throws Exception {
        CompositeResultSurveillance compositeResultSurveillance = new CompositeResultSurveillance();
        compositeResultSurveillance.setFoaSurveillance(itemProcessorSurveillance.process(surveillanceLineFile));
        compositeResultSurveillance.setFoaSurveillanceEcheance(itemProcessorSurveillanceEcheance.process(surveillanceLineFile));
        return compositeResultSurveillance;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return null;
    }
}

The ItemProcessorSurveillance :

@Slf4j
@Component
public class ItemProcessorSurveillance implements ItemProcessor<SurveillanceLineFile, FoaSurveillance>, StepExecutionListener {

    String fileName;

    private StepExecution stepExecution;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        return ExitStatus.COMPLETED;
    }

    @Override
    public FoaSurveillance process(SurveillanceLineFile surveillanceLineFile) throws Exception {
        ExecutionContext executionContext = stepExecution.getExecutionContext();
        
        // Process Data
    }

And the CompositeResult returned by the Processor :

@Getter
@Setter
public class CompositeResultSurveillance {

    private FoaSurveillance foaSurveillance;
    private FoaSurveillanceEcheance foaSurveillanceEcheance;
}

For now I've got a NPE on the ItemProcessorSurveillance because stepExecution is null on the process method.

I can't figure out what's wrong. Any help ?

Answer 1

This is because your ItemProcessorSurveillance implements two interfaces: ItemProcessor and StepExecutionListener but is only registered as an ItemProcessor in the step. It should also be registered as a listener so that beforeStep is called when appropriate to set the stepExecution field.