Add ability for Task to obtain next Job Params for job launch

Update tests so that they can handle that a JobIncrementer is fired on job launch if one exists and update tests.
spring-projects/spring-batch#4910

Author: cppwfs

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunner.java

@@ -30,9 +30,9 @@
 import org.springframework.batch.core.job.Job;
 import org.springframework.batch.core.job.JobExecution;
 import org.springframework.batch.core.job.JobExecutionException;
+import org.springframework.batch.core.job.JobInstance;
 import org.springframework.batch.core.job.parameters.JobParameter;
 import org.springframework.batch.core.job.parameters.JobParameters;
-import org.springframework.batch.core.job.parameters.JobParametersBuilder;
 import org.springframework.batch.core.job.parameters.JobParametersIncrementer;
 import org.springframework.batch.core.job.parameters.JobParametersInvalidException;
 import org.springframework.batch.core.launch.JobOperator;
@@ -50,6 +50,7 @@
 import org.springframework.cloud.task.listener.TaskException;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -90,7 +91,7 @@ public class TaskJobLauncherApplicationRunner extends JobLauncherApplicationRunn
 	 */
 	public TaskJobLauncherApplicationRunner(JobOperator jobOperator, JobRepository jobRepository,
 			TaskBatchProperties taskBatchProperties) {
-		super(jobOperator, jobRepository);
+		super(jobOperator);
 		this.taskJobOperator = jobOperator;
 		this.taskJobRepository = jobRepository;
 		this.taskBatchProperties = taskBatchProperties;
@@ -136,9 +137,12 @@ protected void execute(Job job, JobParameters jobParameters)
 		else {
 			JobParametersIncrementer incrementer = job.getJobParametersIncrementer();
 			if (incrementer != null) {
-				JobParameters nextParameters = new JobParametersBuilder(jobParameters, this.taskJobRepository)
-					.getNextJobParameters(job)
-					.toJobParameters();
+				// JobParameters nextParameters = new JobParametersBuilder(jobParameters,
+				// this.taskJobRepository)
+				// .getNextJobParameters(job)
+				// .toJobParameters();
+				JobParameters nextParameters = getNextJobParameters(job, new HashMap<>(jobParameters.getParameters()),
+						this.taskJobRepository);
 				parameters = merge(nextParameters, jobParameters);
 			}
 		}
@@ -223,4 +227,51 @@ private JobParameters merge(JobParameters parameters, JobParameters additionals)
 		return new JobParameters(merged);
 	}
 
+	/**
+	 * Initializes the {@link JobParameters} based on the state of the {@link Job}. This
+	 * should be called after all parameters have been entered into the builder. All
+	 * parameters already set on this builder instance are appended to those retrieved
+	 * from the job incrementer, overriding any with the same key (this is the same
+	 * behavior as
+	 * {@link org.springframework.batch.core.launch.support.CommandLineJobRunner} with the
+	 * {@code -next} option and
+	 * {@link org.springframework.batch.core.launch.JobOperator#startNextInstance(String)}).
+	 * @param job The job for which the {@link JobParameters} are being constructed.
+	 * @return a reference to this object.
+	 *
+	 * @since 4.0
+	 */
+	public JobParameters getNextJobParameters(Job job, Map<String, JobParameter<?>> parameterMap,
+			JobRepository taskJobRepository) {
+		Assert.notNull(job, "Job must not be null");
+		Assert.notNull(job.getJobParametersIncrementer(),
+				"No job parameters incrementer found for job=" + job.getName());
+
+		String name = job.getName();
+		JobParameters nextParameters;
+		JobInstance lastInstance = taskJobRepository.getLastJobInstance(name);
+		JobParametersIncrementer incrementer = job.getJobParametersIncrementer();
+		if (lastInstance == null) {
+			// Start from a completely clean sheet
+			nextParameters = incrementer.getNext(new JobParameters());
+		}
+		else {
+			JobExecution previousExecution = taskJobRepository.getLastJobExecution(lastInstance);
+			if (previousExecution == null) {
+				// Normally this will not happen - an instance exists with no executions
+				nextParameters = incrementer.getNext(new JobParameters());
+			}
+			else {
+				nextParameters = incrementer.getNext(previousExecution.getJobParameters());
+			}
+		}
+
+		// start with parameters from the incrementer
+		Map<String, JobParameter<?>> nextParametersMap = new HashMap<>(nextParameters.getParameters());
+		// append new parameters (overriding those with the same key)
+		nextParametersMap.putAll(parameterMap);
+		parameterMap = nextParametersMap;
+		return new JobParameters(parameterMap);
+	}
+
 }

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunnerCoreTests.java

@@ -35,7 +35,6 @@
 import org.springframework.batch.core.launch.JobOperator;
 import org.springframework.batch.core.launch.support.RunIdIncrementer;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.core.step.Step;
 import org.springframework.batch.core.step.builder.StepBuilder;
 import org.springframework.batch.core.step.tasklet.Tasklet;
@@ -124,12 +123,15 @@ void retryFailedExecutionOnNonRestartableJob() {
 			// A failed job that is not restartable does not re-use the job params of
 			// the last execution, but creates a new job instance when running it again.
 			assertThat(jobLauncherContext.jobInstances()).hasSize(2);
-			assertThatExceptionOfType(JobRestartException.class).isThrownBy(() -> {
+			assertThatExceptionOfType(TaskException.class).isThrownBy(() -> {
 				// try to re-run a failed execution
+				// In this case the change from the previous behavior is that a new job
+				// instance is created
+				// https://github.com/spring-projects/spring-batch/issues/4910
 				jobLauncherContext.runner.execute(job,
 						new JobParametersBuilder().addLong("run.id", 1L).toJobParameters());
-				fail("expected JobRestartException");
-			}).withMessageContaining("JobInstance already exists and is not restartable");
+				fail("expected TaskException");
+			}).withMessageContaining("Job job failed during execution for job instance id 3 with jobExecutionId of 3 ");
 		});
 	}
 
@@ -148,9 +150,11 @@ void retryFailedExecutionWithNonIdentifyingParameters() {
 			runFailedJob(jobLauncherContext, job, jobParameters);
 			assertThat(jobLauncherContext.jobInstances()).hasSize(1);
 			// try to re-run a failed execution with non identifying parameters
+			// Updated to expect a new 2 instances are created because of this change
+			// https://github.com/spring-projects/spring-batch/issues/4910
 			runFailedJob(jobLauncherContext, job,
 					new JobParametersBuilder(jobParameters).addLong("run.id", 1L).toJobParameters());
-			assertThat(jobLauncherContext.jobInstances()).hasSize(1);
+			assertThat(jobLauncherContext.jobInstances()).hasSize(2);
 		});
 	}