Error handling in multithreaded Spring Batch

Question

I have developed a Spring Batch application that works fine with single thread. Its a simple batch application that reads a csv file using FlatFileItemReader outputs a POJO CSVLineMapper, does simple processing and then write the POJO to a repository.

Now I make the application multithreaded using ThreadPoolTaskExecutor. To test the error handling by the framework, I throw RuntimeException for a specific record in the processor expecting only hte specific thread to get terminated and skipping only that chunk in which error was thrown. But the application terminates after the error writing only 15 records. why? Am I doing something wrong?

As restartability is not supported with multithreading, how do we design a multi-threaded spring batch application such that only problematic record is skipped and the application continues processing hte other records without terminating.

Please find the code snippet used below :

public Step load(){
ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
threadPoolTaskExecutor.setMaxPoolSize(5);
threadPoolTaskExecutor.setCorePoolSize(5);
threadPoolTaskExecutor.afterPropertiesSet();
stepBuilderFactory.get("load")
        .chunk(5)
        .reader(reader1())
        .processor(processor())
        .writer(writer())
        .taskExecutor(threadPoolTaskExecutor)
        .listener(stepExecutionListener)
        .listener(processListener())
        .listener(writeListener)
        .build()
}

reader1 is FlatFileItemReader that has setSaveState to false.

One more observation is that the log in the reader is called only once in the complete flow which is called my the main thread. But the processor and writer are called by diff threads of ThreadPoolTaskExecutor. Why? The reader doesnot implement ItemReader, but processor and writer implement ItemProcessor and ItemWriter respectively in my case.

Answer 1

The sysout in the reader prints the current thread as main thread. But the reader should be part of a thread in ThreadPoolExecutor?

Because you added the System.out.println in the method that defines the Spring bean in the application context. The Spring application context is configured by the main thread. At that point, your job is not started yet and the reader is not invoked yet. In a multi-threaded step, the reader should also be called by one of the treads of the task executor, you can see it by adding some logging in the read method for example:

@Bean
@StepScope
public FlatFileItemReader<Person> reader1() {
    FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>() {
        @Override
        public Person read() throws Exception {
            System.out.println("Thread running when invoking the reader: " + Thread.currentThread().getName());
            return super.read();
        }
    };
    reader.setLinesToSkip(1);
    reader.setResource(new FileSystemResource("persons.csv"));

    DefaultLineMapper<Person> lineMapper = new DefaultLineMapper<>();
    DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();

    String[] names = {"firstname", "lastname", "age"};
    tokenizer.setNames(names);

    BeanWrapperFieldSetMapper<Person> fieldSetMapper = new BeanWrapperFieldSetMapper<>();
    fieldSetMapper.setTargetType(Person.class);

    lineMapper.setLineTokenizer(tokenizer);
    lineMapper.setFieldSetMapper(fieldSetMapper);
    lineMapper.afterPropertiesSet();

    reader.setLineMapper(lineMapper);
    reader.setSaveState(false);
    System.out.println("Thread running when configuring the reader bean: " + Thread.currentThread().getName());
    return reader;

}

With your example, this prints:

Thread running when configuring the reader bean: main
Thread running when invoking the reader: ThreadPoolTaskExecutor-3
Thread running when invoking the reader: ThreadPoolTaskExecutor-2
Thread running when invoking the reader: ThreadPoolTaskExecutor-1
Thread running when invoking the reader: ThreadPoolTaskExecutor-2
Thread running when invoking the reader: ThreadPoolTaskExecutor-3
Thread running when invoking the reader: ThreadPoolTaskExecutor-1

Why the application is terminating without completing all the records?

Because your step is not a fault-tolerant step. A simple chunk-oriented step fails at the first error. You can turn your step into a fault-tolerant one a configure your retry/skip logic as needed, something like:

Step step = this.stepBuilderFactory.get("load")
                .<Person, Person1>chunk(2)
                .reader(reader1())
                .processor(processor1())
                .writer(writer())
                .faultTolerant()
                .skipPolicy(new YourSkipPoilcy())
                .retryPolicy(new YourRetryPolicy)
                .taskExecutor(threadPoolTaskExecutor)
                .build();

Please refer to the "Configuring skip logic" and "Configuring the retry logic" sections in the reference documentation

Answer 2

I wrote a mimimal poc to reproduce the problem. What is happening is , the application terminates after 15-16 records. It never writes the rest of the records in csv. Person and Person1 are pojos with firstname,lastname,age as fields.

Person.csv has columns firstname,lastname,age. Issue comes when more than 20 records. Have one record, lets say the 8th record with firstname "zxcv" (problematic record).

The sysout in the reader prints the current thread as main thread. But the reader should be part of a thread in ThreadPoolExecutor? Why the application is terminating without completing all the records?

Please find code below:

@SpringBootApplication
@EnableBatchProcessing
public class DemoApplication implements ApplicationRunner {

    private JobLauncher jobLauncher;

    private static ConfigurableApplicationContext ctx;
    static DemoApplication _this;

    public static void main(String...args) {
            try {
                ctx = SpringApplication.run(DemoApplication.class, args);
                _this.start(args);
                System.exit(0);
            }catch(Exception e){
                System.exit(-1);
            }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        _this=this;
    }

    public void start(String...args) throws JobInstanceAlreadyCompleteException, JobExecutionAlreadyRunningException, JobParametersInvalidException, JobRestartException {
        JobParameters jobParameters = new JobParametersBuilder()
                .addString("user", args[0])
                .addString("repository", args[1])
                .toJobParameters();
        this.jobLauncher.run((Job) ctx.getBean("uploadJob"),jobParameters);
    }
}




@Configuration
public class BulkConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    @StepScope
    public FlatFileItemReader<Person> reader1(){
        FlatFileItemReader<Person>  reader = new FlatFileItemReader<>();
        reader.setLinesToSkip(1);
        reader.setResource(new PathResource("C:\\sush\\setups\\demo\\src\\main\\resources\\person.csv"));

        DefaultLineMapper<Person> lineMapper = new DefaultLineMapper<>();
        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();

        String[] names = {"firstname", "lastname", "age"};
        tokenizer.setNames(names);

        BeanWrapperFieldSetMapper<Person>  fieldSetMapper = new BeanWrapperFieldSetMapper<>();
        fieldSetMapper.setTargetType(Person.class);

        lineMapper.setLineTokenizer(tokenizer);
        lineMapper.setFieldSetMapper(fieldSetMapper);
        lineMapper.afterPropertiesSet();

        reader.setLineMapper(lineMapper);
        reader.setSaveState(false);
        System.out.println("Thread running in reader "+ Thread.currentThread().getName());
        return reader;

    }

    @Bean
    @StepScope
    public ItemProcessor<Person,Person1> processor1(){
        return new EnrichmentProcessor();
    }

    @Bean
    @StepScope
    public ItemWriter<Person1> writer(){
        return new FileUploader();
    }


    @Bean
    public Step load(){
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setMaxPoolSize(3);
        threadPoolTaskExecutor.setCorePoolSize(3);
        threadPoolTaskExecutor.afterPropertiesSet();

        return this.stepBuilderFactory.get("load")
                .<Person,Person1>chunk(2)
                .reader(reader1())
                .processor(processor1())
                .writer(writer())
                .taskExecutor(threadPoolTaskExecutor)
                .build();
    }


    @Bean(name = "uploadJob")
    public Job cmUploadJob(){
        return this.jobBuilderFactory.get("uploadJob")
                .start(load())
                .build();
    }
}



public class EnrichmentProcessor implements ItemProcessor<Person, Person1> {
    @Override
    public Person1 process(Person person) throws Exception {
        Person1 person1 = new Person1();
        person1.setFirstname(person.getFirstname().toUpperCase());
        person1.setLastname(person.getLastname().toUpperCase());
        person1.setAge(person.getAge());

        if(person.getFirstname().equals("zxcv")){
            throw new RuntimeException("Error occurred in thread " + Thread.currentThread().getName());
        }

        return person1;
    }
}


public class FileUploader implements ItemWriter<Person1> {
    @Override
    public void write(List<? extends Person1> list) throws Exception {
        for(Person1 p : list){
            System.out.println("Writing person "+ p.getFirstname() + Thread.currentThread().getName());
        }
    }
}