Spring Batch requires unique job parameters for its execution.so you can add the current time as a job parameter

Map<String, JobParameter> confMap = new HashMap<String, JobParameter>();
confMap.put("time", new JobParameter(System.currentTimeMillis()));
JobParameters jobParameters = new JobParameters(confMap);
jobLauncher.run(springCoreJob, jobParameters);

We had the same issue and to me this looks like a flaw, especially when you compare how the SimpleJobLauncher evaluates the last job.

As you mentioned, getLasFailedJobExecution() does not take into account the parameters. (Which, at least in my opinion, is wrong. You can run the same job with different identifying parameters, but restarting would only work if only one of this runs failed.)

On the other hand, SimpleJobLauncher gets the last jobexecution based on the parameters:

public JobExecution run(final Job job, final JobParameters jobParameters)
        throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException,
        JobParametersInvalidException {
...

    JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
...

This is not consistent.

The way we fixed it was to implement our own CommandLineJobRunner that is derived from the CommandLineJobRunner, but you have to overwrite the whole start method and, therefore, you alse will have to copy a couple of private methods.

The whole restart part in the start method has to be changed in something like this:

        if (opts.contains("-restart")) {
            JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);

            if (!jobExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) {
                throw new JobExecutionNotFailedException("No failed or stopped execution found for job="
                        + jobIdentifier);
            }
            jobParameters = jobExecution.getJobParameters();
            jobName = jobExecution.getJobInstance().getJobName();
        }

Edited

Sometimes I really should read the javadoc...

In case of a restart, you could pass the executionid instead of the jobname. In this case, it will find the right execution to restart.

From the javadoch of the main method:

restart: (optional) if the job has failed or stopped and the most should be restarted. If specified then the jobIdentifier parameter can be interpreted either as the name of the job or the id of the job execution that failed.

Therefore, using the execution id in case of a restart will ensure that the right instance is restarted. In this case, you also don't have to provide any other parameters, since they will be taken from the last execution.

Going by the concept of Job Instances and Job Executions in Spring Batch, you can't start a COMPLETED job instance again though you can launch same instance of job again & again till its not COMPLETE ( and few more job statuses ).

Job instance uniqueness is achieved by jobId & job parameters.

So if you don't vary your parameters, you can't start a completed job with same instance.

If your REST End Point is restricted to take same value again and again, you need to add a unique parameter within that method and that can very well be - java.lang.System.currentTimeMillis().

You can use any other unique value though current system time is very convenient.

Convert java.lang.System.currentTimeMillis() to String and append to your parameters with something like , String.format("paymentPeriod=%s", paymentPeriod+java.lang.System.currentTimeMillis()) instead of what you are currently.

Hope you get the idea.

Just to recap what was actually done based on the advice provided by incomplete-co.de. I created a recovery flow which is similar to the one below. The recovery flow wraps my actual batch and responsible only to serve the correct job parameters to the internal job. It could be initial parameters on first execution, new parameters on normal execution or old parameters in case the last execution had failed.

<batch:job id="recoveryWrapper"
       incrementer="wrapperRunIdIncrementer"
       restartable="true">
    <batch:decision id="recoveryFlowDecision" decider="recoveryFlowDecider">
        <batch:next on="FIRST_RUN" to="defineParametersOnFirstRun" />
        <batch:next on="RECOVER" to="recover.batchJob " />
        <batch:next on="CURRENT" to="current.batchJob " />
    </batch:decision>
    <batch:step id="defineParametersOnFirstRun" next="current.batchJob">
        <batch:tasklet ref="defineParametersOnFirstRunTasklet"/>
    </batch:step>
    <batch:step id="recover.batchJob " next="current.batchJob">
        <batch:job ref="batchJob" job-launcher="jobLauncher"
                            job-parameters-extractor="jobParametersExtractor" />
    </batch:step>
    <batch:step id="current.batchJob" >
       <batch:job ref="batchJob" job-launcher="jobLauncher"
                            job-parameters-extractor="jobParametersExtractor" />
    </batch:step>
</batch:job>

The heart of the solution is the RecoveryFlowDecider the JobParametersExtractor while using Spring Batch Restart mechanism. RecoveryFlowDecider will query the JobExplorer and JobRepository to find out if we had a failure in the last run. It will place The last execution on the execution context of the wrapper to use later in the JobParametersExtractor. Note the use of runIdIncremeter to allow re-execution of the wrapper job.

@Component
 public class RecoveryFlowDecider implements JobExecutionDecider {
        private static final String FIRST_RUN = "FIRST_RUN";
        private static final String CURRENT = "CURRENT";
        private static final String RECOVER = "RECOVER";

        @Autowired
        private JobExplorer jobExplorer;
        @Autowired
        private JobRepository jobRepository;

        @Override
        public FlowExecutionStatus decide(JobExecution jobExecution
                                         ,StepExecution stepExecution) {
            // the wrapper is named as the wrapped job + WRAPPER
            String wrapperJobName = jobExecution.getJobInstance().getJobName();
            String jobName;
             jobName = wrapperJobName.substring(0,wrapperJobName.indexOf(EtlConstants.WRAPPER));
            List<JobInstance> instances = jobExplorer.getJobInstances(jobName, 0, 1);
                JobInstance internalJobInstance = instances.size() > 0 ? instances.get(0) : null;

            if (null == internalJobInstance) {
                return new FlowExecutionStatus(FIRST_RUN);
            }
            JobExecution lastExecution = jobRepository.getLastJobExecution(internalJobInstance.getJobName()
    ,internalJobInstance.getJobParameters());
            //place the last execution on the context (wrapper context to use later)
            jobExecution.getExecutionContext().put(EtlConstants.LAST_EXECUTION, lastExecution);
                ExitStatus exitStatus = lastExecution.getExitStatus();

            if (ExitStatus.FAILED.equals(exitStatus) || ExitStatus.UNKNOWN.equals(exitStatus)) {
                return new FlowExecutionStatus(RECOVER);
            }else if(ExitStatus.COMPLETED.equals(exitStatus)){
                return new FlowExecutionStatus(CURRENT);    
            }
            //We should never get here unless we have a defect
            throw new RuntimeException("Unexpecded batch status: "+exitStatus+" in decider!");

        }

}

Then the JobParametersExtractor will test again for the outcome of the last execution, in case of failed job it will serve the original parameters used to execute the failed job triggering Spring Bacth restart mechanism. Otherwise it will create a new set of parameters and will execute at his normal course.

    @Component
    public class JobExecutionWindowParametersExtractor implements
            JobParametersExtractor {
        @Override
        public JobParameters getJobParameters(Job job, StepExecution stepExecution) {

            // Read the last execution from the wrapping job
            // in order to build Next Execution Window
            JobExecution lastExecution= (JobExecution) stepExecution.getJobExecution().getExecutionContext().get(EtlConstants.LAST_EXECUTION);;

            if(null!=lastExecution){
                if (ExitStatus.FAILED.equals(lastExecution.getExitStatus())) {
                    JobInstance instance = lastExecution.getJobInstance();
                    JobParameters parameters = instance.getJobParameters();                 
                                return parameters;
                }
            }       
            //We do not have failed execution or have no execution at all we need to create a new execution window
            return buildJobParamaters(lastExecution,stepExecution);
        }
...
}

It seems you have to configure the following beans to be able to restart your job.

@Bean
public JobOperator jobOperator(final JobLauncher jobLauncher, final JobRepository jobRepository,
        final JobRegistry jobRegistry, final JobExplorer jobExplorer) {
    final SimpleJobOperator jobOperator = new SimpleJobOperator();
    jobOperator.setJobLauncher(jobLauncher);
    jobOperator.setJobRepository(jobRepository);
    jobOperator.setJobRegistry(jobRegistry);
    jobOperator.setJobExplorer(jobExplorer);
    return jobOperator;
}

@Bean
public JobExplorer jobExplorer(final DataSource dataSource) throws Exception {
    final JobExplorerFactoryBean bean = new JobExplorerFactoryBean();
    bean.setDataSource(dataSource);
    bean.setTablePrefix("BATCH_");
    bean.setJdbcOperations(new JdbcTemplate(dataSource));
    bean.afterPropertiesSet();
    return bean.getObject();
}

Then you need to retrieve the batch instance id from the batch tables to be able to restart that specific instance by using the jobOperator.

final Long restartId = jobOperator.restart(id);
final JobExecution restartExecution = jobExplorer.getJobExecution(restartId);

TL;DR: Spring Batch will always create new Job Execution and will not reuse a previous failed job execution to continue its execution.

Longer answer: First you need to understand three similar but different concept in Spring Batch: Job, Job Instance, Job Execution

I always use this example:

• Job : End-Of-Day Batch
• Job Instance : End-Of-Day Batch for 2018-01-01
• Job Execution: End-Of-Day Batch for 2018-01-01, execution #1

In high-level, that's how Spring Batch's recovery works:

Assuming your first execution failed in the step 3. You can submit the same Job (End-of-Day Batch) with same Parameters (2018-01-01). Spring Batch will try to look up last Job Execution (End-Of-Day Batch for 2018-01-01, execution #1) of the submitted Job Instance (End-of-Day Batch for 2018-01-01), and found that it has previously failed in step 3. Spring Batch will then create a NEW execution, [End-Of-Day Batch for 2018-01-01, execution #2], and start the execution from step 3.

So by design, what Spring trying to recover is a previously failed Job Instance (instead of Job Execution). Spring batch will not reuse execution when you are re-running a previous-failed execution.

In Spring Batch, a job instance is identified by the (identifying) job parameters. Please check the The domain language of Batch section to understand the difference between the Job, JobInstance and JobExecution concepts and how parameters are used to identify job instances.

I tried to stop the batch middle of process and run again. It always executing a new job.

In your case, since your are adding the current time as a job parameter on each run here:

JobParametersBuilder paramsBuilder = new JobParametersBuilder();
paramsBuilder.addString("date", LocalDateTime.now().toString());

you end up with a different job instance each time. If you want to start the same job instance again, you need to pass the same timestamp of the first attempt as a job parameter.