FlatfileItemWriter with Compositewriter example

Question

I have a spring batch that will read the CSV file, then process and write it to another CSV file. I want to write the results into two different flat files based on the process result. Need to write successfully processed records in one file and failure records in another file.

I saw few examples to use "CompositeItemWriter" but there's no exact example for multiple "FlatfileItemWriters".

Anyone, please share the example for my use case?

Answer 1

CompositeItemWriter is used when valid items should be written to two or more outputs. In your case, a SkipListener is more appropriate and can be used to write invalid items to a different file. Here is a quick example:

import java.util.Arrays;
import java.util.Collections;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.listener.SkipListenerSupport;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.file.transform.PassThroughLineAggregator;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;

@Configuration
@EnableBatchProcessing
public class SO65996526 {

    @Bean
    public ItemReader<Integer> itemReader() {
        return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
    }

    @Bean
    public ItemProcessor<Integer, Integer> itemProcessor() {
        return item -> {
            if (item.equals(3)) {
                throw new IllegalArgumentException("no 3 here!");
            }
            if (item.equals(7)) {
                throw new IllegalArgumentException("no 7 here!");
            }
            return item;
        };
    }

    @Bean
    public ItemWriter<Integer> itemWriter() {
        return new FlatFileItemWriterBuilder<Integer>()
                .name("itemWriter")
                .resource(new FileSystemResource("output.txt"))
                .lineAggregator(new PassThroughLineAggregator<>())
                .build();
    }
    
    @Bean
    public FlatFileItemWriter<Integer> skippedItemWriter() {
        return new FlatFileItemWriterBuilder<Integer>()
                .name("skippedItemWriter")
                .resource(new FileSystemResource("skipped.txt"))
                .lineAggregator(new PassThroughLineAggregator<>())
                .build();
    }

    @Bean
    public Job job(JobBuilderFactory jobs, StepBuilderFactory steps) {
        return jobs.get("job")
                .start(steps.get("step")
                        .<Integer, Integer>chunk(5)
                        .reader(itemReader())
                        .processor(itemProcessor())
                        .writer(itemWriter())
                        .faultTolerant()
                        .skip(IllegalArgumentException.class)
                        .skipLimit(3)
                        .listener(new SkippedItemsListener(skippedItemWriter()))
                        .stream(skippedItemWriter()) // ensure open/close are called properly
                        .build())
                .build();
    }

    static class SkippedItemsListener extends SkipListenerSupport<Integer, Integer> {
        private FlatFileItemWriter<Integer> skippedItemsWriter;

        public SkippedItemsListener(FlatFileItemWriter<Integer> skippedItemsWriter) {
            this.skippedItemsWriter = skippedItemsWriter;
        }

        @Override
        public void onSkipInProcess(Integer item, Throwable t) {
            try {
                skippedItemsWriter.write(Collections.singletonList(item));
            } catch (Exception e) {
                Logger.getLogger(this.getClass().getName()).log(Level.SEVERE, "Unable to write skipped item " + item);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(SO65996526.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

}

This example skips items 3 and 7 and produces two files: output.txt with valid items and skipped.txt with invalid items.