job restart does not restart steps with previously saved contexts

[heerdev]

I have a job which saved the pojo to database. For each object I am saving the the unique key of the data as execution context in step. I have implemented the reader with ItemStream and using update method to store the custom execution context.

Now, When I am restarting the FAILED job instance using simpleJobOperator.restart(JOB_EXECUTION_ID), the job restarts, but it cannot get the saved custom exeuction context of previosuly FAILED step.

As a work around, I have to save the custom execution context as job execution context.

[mminella]

Can you please provide the code for your ItemReader and its configuration?

[heerdev]

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.*;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.core.io.FileSystemResource;
import org.springframework.core.io.Resource;
import za.co.momentum.mmibatch.domain.Customer;
import za.co.momentum.mmibatch.domain.CustomerFieldSetMapper;

public class ChunkFileItemReader implements ItemReader, ItemStream {
Logger log = LoggerFactory.getLogger(ChunkFileItemReader.class);
private Resource resource ;
private FlatFileItemReader itemReader;
private String currentName;
private String restartName;
private static final String CURRENT_INDEX="name";

@Override
public Customer read() throws Exception, ParseException, NonTransientResourceException {


    Customer customer = itemReader.read();

    if(customer!=null && restartName!=null){
        if(restartName.equalsIgnoreCase(customer.getName())){
            return null;
        }
        currentName = customer.getName();
    }
    log.info("READ  {}", customer!=null ? customer.getName() : null);
    return customer;
}



@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
             executionContext.put(CURRENT_INDEX,currentName);
      }

@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
    itemReader = new FlatFileItemReader<>();
    this.resource = new FileSystemResource("C:\\project\\example\\spring-integration-sample\\sping-batch\\src\\main\\resources\\data\\customerdata.csv");
    itemReader.setResource(this.resource);
    itemReader.open(executionContext);
    DefaultLineMapper<Customer> lineMapper = new DefaultLineMapper<>();
    lineMapper.setLineTokenizer(new DelimitedLineTokenizer());
    lineMapper.setFieldSetMapper( new CustomerFieldSetMapper());
    itemReader.setLineMapper(lineMapper);

}

@Override
public void close() throws ItemStreamException {
    itemReader.close();
}

}

[heerdev]

//Configuration class

@configuration
@EnableBatchProcessing
class SpringBatchConfig {

private final JobBuilderFactory jobs;
private final  StepBuilderFactory steps;

public SpringBatchConfig(JobBuilderFactory jobs,StepBuilderFactory steps){
    this.jobs=jobs;
    this.steps = steps;
}

@Bean
public ItemReader<Customer> flatFileItemReader(){
    return new ChunkFileItemReader();
}

@Bean
public ItemWriter<Customer> customerItemWriter(){
    return new ChunkItemWriter();
}

@Bean
public Step readDateFromFile(){
   return this.steps.get("readDateFromFile")
           .<Customer,Customer>chunk(2)
           .reader(flatFileItemReader())
           .writer(customerItemWriter())
           //.taskExecutor(taskExecutor())
           .build();
}

@Bean("readFileJob")
public Job readFileJob(){
    return this.jobs.get("readFileJob")
            .start(readDateFromFile())
            .build();
}


@Bean
public TaskExecutor taskExecutor(){
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("MMI-Batch-");
    //taskExecutor.setConcurrencyLimit(10);
    return taskExecutor;
}

@Bean
public SimpleJobOperator jobOperator(JobExplorer jobExplorer,
                                     JobRepository jobRepository,
                                     JobRegistry jobRegistry) throws DuplicateJobException {
    ReferenceJobFactory referenceJobFactory = new ReferenceJobFactory(readFileJob());

    SimpleJobOperator jobOperator = new SimpleJobOperator();
    jobRegistry.register(referenceJobFactory);
    jobOperator.setJobExplorer(jobExplorer);
    jobOperator.setJobRepository(jobRepository);
    jobOperator.setJobRegistry(jobRegistry);
    jobOperator.setJobLauncher(simpleJobLauncher(jobRepository));
    return jobOperator;
}

@Bean
JobLauncher simpleJobLauncher(JobRepository jobRepository){
    SimpleJobLauncher  simpleJobLauncher = new SimpleJobLauncher();
    simpleJobLauncher.setJobRepository(jobRepository);
    return simpleJobLauncher;
}

[mminella]

In your open method, you are not resetting the reader's currentName field. The ItemStream#open is the location for resetting the state on a restart. Also, these questions are best handled via StackOverflow (the leads of the Spring projects monitor the project tags there). I'm closing this issue. If you have further questions about this, please create a question on StackOverflow where we can discuss it in more depth.

[heerdev]

It worked, thanks :)