Tasks supports JobLauncherApplicationRunner

Will  support JobLauncherCommandLineRunner  if running task on Boot 2.2.0

resolves #645

Updated based on code review and rebased

Updated code based on code review

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherApplicationRunnerFactoryBean.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.batch.configuration;
+
+import java.util.List;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.configuration.JobRegistry;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.boot.autoconfigure.batch.BatchProperties;
+import org.springframework.cloud.task.batch.handler.TaskJobLauncherApplicationRunner;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * Factory bean for creating an instance of {@link TaskJobLauncherApplicationRunner}.
+ *
+ * @author Glenn Renfro
+ * @since 2.3.0
+ */
+public class TaskJobLauncherApplicationRunnerFactoryBean
+		implements FactoryBean<TaskJobLauncherApplicationRunner> {
+
+	private JobLauncher jobLauncher;
+
+	private JobExplorer jobExplorer;
+
+	private List<Job> jobs;
+
+	private String jobNames;
+
+	private JobRegistry jobRegistry;
+
+	private Integer order = 0;
+
+	private TaskBatchProperties taskBatchProperties;
+
+	private JobRepository jobRepository;
+
+	public TaskJobLauncherApplicationRunnerFactoryBean(JobLauncher jobLauncher,
+			JobExplorer jobExplorer, List<Job> jobs,
+			TaskBatchProperties taskBatchProperties, JobRegistry jobRegistry,
+			JobRepository jobRepository, BatchProperties batchProperties) {
+		Assert.notNull(taskBatchProperties, "taskBatchProperties must not be null");
+		Assert.notNull(batchProperties, "batchProperties must not be null");
+		Assert.notEmpty(jobs, "jobs must not be null nor empty");
+
+		this.jobLauncher = jobLauncher;
+		this.jobExplorer = jobExplorer;
+		this.jobs = jobs;
+		this.jobNames = taskBatchProperties.getJobNames();
+		this.jobRegistry = jobRegistry;
+		this.taskBatchProperties = taskBatchProperties;
+		if (StringUtils.hasText(batchProperties.getJob().getNames())) {
+			this.jobNames = batchProperties.getJob().getNames();
+		}
+		else {
+			this.jobNames = taskBatchProperties.getJobNames();
+		}
+		this.order = taskBatchProperties.getCommandLineRunnerOrder();
+		this.jobRepository = jobRepository;
+	}
+
+	public void setOrder(int order) {
+		this.order = order;
+	}
+
+	@Override
+	public TaskJobLauncherApplicationRunner getObject() {
+		TaskJobLauncherApplicationRunner taskJobLauncherApplicationRunner = new TaskJobLauncherApplicationRunner(
+				this.jobLauncher, this.jobExplorer, this.jobRepository,
+				this.taskBatchProperties);
+		taskJobLauncherApplicationRunner.setJobs(this.jobs);
+		if (StringUtils.hasText(this.jobNames)) {
+			taskJobLauncherApplicationRunner.setJobNames(this.jobNames);
+		}
+		taskJobLauncherApplicationRunner.setJobRegistry(this.jobRegistry);
+
+		if (this.order != null) {
+			taskJobLauncherApplicationRunner.setOrder(this.order);
+		}
+		return taskJobLauncherApplicationRunner;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return TaskJobLauncherApplicationRunner.class;
+	}
+
+}

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,6 +27,8 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
 import org.springframework.boot.autoconfigure.batch.BatchProperties;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
@@ -48,6 +50,7 @@ public class TaskJobLauncherAutoConfiguration {
 	private TaskBatchProperties properties;
 
 	@Bean
+	@ConditionalOnMissingClass("org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner")
 	public TaskJobLauncherCommandLineRunnerFactoryBean jobLauncherCommandLineRunner(
 			JobLauncher jobLauncher, JobExplorer jobExplorer, List<Job> jobs,
 			JobRegistry jobRegistry, JobRepository jobRepository,
@@ -60,4 +63,19 @@ public class TaskJobLauncherAutoConfiguration {
 		return taskJobLauncherCommandLineRunnerFactoryBean;
 	}
 
+	@Bean
+	@ConditionalOnClass(
+			name = "org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner")
+	public TaskJobLauncherApplicationRunnerFactoryBean taskJobLauncherApplicationRunner(
+			JobLauncher jobLauncher, JobExplorer jobExplorer, List<Job> jobs,
+			JobRegistry jobRegistry, JobRepository jobRepository,
+			BatchProperties batchProperties) {
+		TaskJobLauncherApplicationRunnerFactoryBean taskJobLauncherApplicationRunnerFactoryBean;
+		taskJobLauncherApplicationRunnerFactoryBean = new TaskJobLauncherApplicationRunnerFactoryBean(
+				jobLauncher, jobExplorer, jobs, this.properties, jobRegistry,
+				jobRepository, batchProperties);
+
+		return taskJobLauncherApplicationRunnerFactoryBean;
+	}
+
 }

spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherCommandLineRunnerFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -33,6 +33,8 @@ import org.springframework.util.StringUtils;
  * Factory bean for creating an instance of {@link TaskJobLauncherCommandLineRunner}.
  *
  * @author Glenn Renfro
+ * @deprecated Use
+ * {@link org.springframework.cloud.task.batch.handler.TaskJobLauncherApplicationRunner}
  */
 public class TaskJobLauncherCommandLineRunnerFactoryBean
 		implements FactoryBean<TaskJobLauncherCommandLineRunner> {

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

@@ -0,0 +1,234 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.task.batch.handler;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.JobParameter;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.JobParametersIncrementer;
+import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.explore.JobExplorer;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.batch.repeat.RepeatStatus;
+import org.springframework.batch.repeat.support.RepeatTemplate;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.batch.JobExecutionEvent;
+import org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner;
+import org.springframework.cloud.task.batch.configuration.TaskBatchProperties;
+import org.springframework.cloud.task.listener.TaskException;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.util.StringUtils;
+
+/**
+ * {@link ApplicationRunner} to {@link JobLauncher launch} Spring Batch jobs. Runs all
+ * jobs in the surrounding context by default and throws an exception upon the first job
+ * that returns an {@link BatchStatus} of FAILED if a {@link TaskExecutor} in the
+ * {@link JobLauncher} is not specified. If a {@link TaskExecutor} is specified in the
+ * {@link JobLauncher} then all Jobs are launched and an exception is thrown if one or
+ * more of the jobs has an {@link BatchStatus} of FAILED. TaskJobLauncherApplicationRunner
+ * can also be used to launch a specific job by providing a jobName. The
+ * TaskJobLauncherApplicationRunner takes the place of the
+ * {@link JobLauncherApplicationRunner} when it is in use.
+ *
+ * @author Glenn Renfro
+ * @since 2.3.0
+ */
+public class TaskJobLauncherApplicationRunner extends JobLauncherApplicationRunner {
+
+	private static final Log logger = LogFactory
+			.getLog(TaskJobLauncherApplicationRunner.class);
+
+	private JobLauncher taskJobLauncher;
+
+	private JobExplorer taskJobExplorer;
+
+	private JobRepository taskJobRepository;
+
+	private List<JobExecution> jobExecutionList = new ArrayList<>();
+
+	private ApplicationEventPublisher taskApplicationEventPublisher;
+
+	private TaskBatchProperties taskBatchProperties;
+
+	/**
+	 * Create a new {@link TaskJobLauncherApplicationRunner}.
+	 * @param jobLauncher to launch jobs
+	 * @param jobExplorer to check the job repository for previous executions
+	 * @param jobRepository to check if a job instance exists with the given parameters
+	 * when running a job
+	 * @param taskBatchProperties the properties used to configure the
+	 * taskBatchProperties.
+	 */
+	public TaskJobLauncherApplicationRunner(JobLauncher jobLauncher,
+			JobExplorer jobExplorer, JobRepository jobRepository,
+			TaskBatchProperties taskBatchProperties) {
+		super(jobLauncher, jobExplorer, jobRepository);
+		this.taskJobLauncher = jobLauncher;
+		this.taskJobExplorer = jobExplorer;
+		this.taskJobRepository = jobRepository;
+		this.taskBatchProperties = taskBatchProperties;
+	}
+
+	public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
+		super.setApplicationEventPublisher(publisher);
+		this.taskApplicationEventPublisher = publisher;
+	}
+
+	@Override
+	public void run(String... args) throws JobExecutionException {
+		logger.info("Running default command line with: " + Arrays.asList(args));
+		launchJobFromProperties(StringUtils.splitArrayElementsIntoProperties(args, "="));
+		monitorJobExecutions();
+	}
+
+	protected void execute(Job job, JobParameters jobParameters)
+			throws JobExecutionAlreadyRunningException, JobRestartException,
+			JobInstanceAlreadyCompleteException, JobParametersInvalidException {
+		String jobName = job.getName();
+		JobParameters parameters = jobParameters;
+		boolean jobInstanceExists = this.taskJobRepository.isJobInstanceExists(jobName,
+				parameters);
+		if (jobInstanceExists) {
+			JobExecution lastJobExecution = this.taskJobRepository
+					.getLastJobExecution(jobName, jobParameters);
+			if (lastJobExecution != null && isStoppedOrFailed(lastJobExecution)
+					&& job.isRestartable()) {
+				// Retry a failed or stopped execution with previous parameters
+				JobParameters previousParameters = lastJobExecution.getJobParameters();
+				/*
+				 * remove Non-identifying parameters from the previous execution's
+				 * parameters since there is no way to remove them programmatically. If
+				 * they are required (or need to be modified) on a restart, they need to
+				 * be (re)specified.
+				 */
+				JobParameters previousIdentifyingParameters = removeNonIdentifying(
+						previousParameters);
+				// merge additional parameters with previous ones (overriding those with
+				// the same key)
+				parameters = merge(previousIdentifyingParameters, jobParameters);
+			}
+		}
+		else {
+			JobParametersIncrementer incrementer = job.getJobParametersIncrementer();
+			if (incrementer != null) {
+				JobParameters nextParameters = new JobParametersBuilder(jobParameters,
+						this.taskJobExplorer).getNextJobParameters(job).toJobParameters();
+				parameters = merge(nextParameters, jobParameters);
+			}
+		}
+		JobExecution execution = this.taskJobLauncher.run(job, parameters);
+		if (this.taskApplicationEventPublisher != null) {
+			this.taskApplicationEventPublisher
+					.publishEvent(new JobExecutionEvent(execution));
+		}
+		this.jobExecutionList.add(execution);
+		if (execution.getStatus().equals(BatchStatus.FAILED)) {
+			throwJobFailedException(Collections.singletonList(execution));
+		}
+	}
+
+	private void monitorJobExecutions() {
+		RepeatTemplate template = new RepeatTemplate();
+
+		Date startDate = new Date();
+
+		template.iterate(context -> {
+
+			List<JobExecution> failedJobExecutions = new ArrayList<>();
+			RepeatStatus repeatStatus = RepeatStatus.FINISHED;
+			for (JobExecution jobExecution : this.jobExecutionList) {
+				JobExecution currentJobExecution = this.taskJobExplorer
+						.getJobExecution(jobExecution.getId());
+				BatchStatus batchStatus = currentJobExecution.getStatus();
+				if (batchStatus.isRunning()) {
+					repeatStatus = RepeatStatus.CONTINUABLE;
+				}
+				if (batchStatus.equals(BatchStatus.FAILED)) {
+					failedJobExecutions.add(jobExecution);
+				}
+			}
+			Thread.sleep(this.taskBatchProperties.getFailOnJobFailurePollInterval());
+
+			if (repeatStatus.equals(RepeatStatus.FINISHED)
+					&& failedJobExecutions.size() > 0) {
+				throwJobFailedException(failedJobExecutions);
+			}
+			return repeatStatus;
+		});
+	}
+
+	private void throwJobFailedException(List<JobExecution> failedJobExecutions) {
+		StringBuilder message = new StringBuilder("The following Jobs have failed: \n");
+		for (JobExecution failedJobExecution : failedJobExecutions) {
+			message.append(String.format("Job %s failed during "
+					+ "execution for job instance id %s with jobExecutionId of %s \n",
+					failedJobExecution.getJobInstance().getJobName(),
+					failedJobExecution.getJobId(), failedJobExecution.getId()));
+		}
+
+		logger.error(message);
+
+		throw new TaskException(message.toString());
+
+	}
+
+	private JobParameters removeNonIdentifying(JobParameters parameters) {
+		Map<String, JobParameter> parameterMap = parameters.getParameters();
+		HashMap<String, JobParameter> copy = new HashMap<>(parameterMap);
+
+		for (Map.Entry<String, JobParameter> parameter : copy.entrySet()) {
+			if (!parameter.getValue().isIdentifying()) {
+				parameterMap.remove(parameter.getKey());
+			}
+		}
+
+		return new JobParameters(parameterMap);
+	}
+
+	private boolean isStoppedOrFailed(JobExecution execution) {
+		BatchStatus status = execution.getStatus();
+		return (status == BatchStatus.STOPPED || status == BatchStatus.FAILED);
+	}
+
+	private JobParameters merge(JobParameters parameters, JobParameters additionals) {
+		Map<String, JobParameter> merged = new HashMap<>();
+		merged.putAll(parameters.getParameters());
+		merged.putAll(additionals.getParameters());
+		return new JobParameters(merged);
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -67,6 +67,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Glenn Renfro
  * @since 2.0.0
+ * @deprecated Use {@link TaskJobLauncherApplicationRunner}
  */
 public class TaskJobLauncherCommandLineRunner extends JobLauncherCommandLineRunner {
 

spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfigurationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,13 +18,14 @@ package org.springframework.cloud.task.batch.configuration;
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
 import org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner;
-import org.springframework.boot.autoconfigure.batch.JobLauncherCommandLineRunner;
 import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration;
 import org.springframework.boot.autoconfigure.jdbc.EmbeddedDataSourceConfiguration;
 import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.task.batch.handler.TaskJobLauncherApplicationRunner;
 import org.springframework.cloud.task.batch.handler.TaskJobLauncherCommandLineRunner;
 import org.springframework.cloud.task.batch.listener.TaskBatchExecutionListenerTests;
 import org.springframework.test.util.ReflectionTestUtils;
@@ -37,10 +38,11 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class TaskJobLauncherAutoConfigurationTests {
 
 	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
+			.withConfiguration(AutoConfigurations.of(BatchAutoConfiguration.class,
+					TaskJobLauncherAutoConfiguration.class))
 			.withUserConfiguration(TaskBatchExecutionListenerTests.JobConfiguration.class,
 					PropertyPlaceholderAutoConfiguration.class,
-					EmbeddedDataSourceConfiguration.class, BatchAutoConfiguration.class,
-					TaskJobLauncherAutoConfiguration.class);
+					EmbeddedDataSourceConfiguration.class);
 
 	@Test
 	public void testAutoBuiltDataSourceWithTaskJobLauncherCLR() {
@@ -48,8 +50,8 @@ public class TaskJobLauncherAutoConfigurationTests {
 				.withPropertyValues("spring.cloud.task.batch.fail-on-job-failure=true")
 				.run(context -> {
 					assertThat(context)
-							.hasSingleBean(TaskJobLauncherCommandLineRunner.class);
-					assertThat(context.getBean(TaskJobLauncherCommandLineRunner.class)
+							.hasSingleBean(TaskJobLauncherApplicationRunner.class);
+					assertThat(context.getBean(TaskJobLauncherApplicationRunner.class)
 							.getOrder()).isEqualTo(0);
 				});
 	}
@@ -60,7 +62,7 @@ public class TaskJobLauncherAutoConfigurationTests {
 				.withPropertyValues("spring.cloud.task.batch.fail-on-job-failure=true",
 						"spring.cloud.task.batch.commandLineRunnerOrder=100")
 				.run(context -> {
-					assertThat(context.getBean(TaskJobLauncherCommandLineRunner.class)
+					assertThat(context.getBean(TaskJobLauncherApplicationRunner.class)
 							.getOrder()).isEqualTo(100);
 				});
 	}
@@ -88,10 +90,10 @@ public class TaskJobLauncherAutoConfigurationTests {
 
 	private void validateJobNames(AssertableApplicationContext context, String jobNames)
 			throws Exception {
-		JobLauncherCommandLineRunner jobLauncherCommandLineRunner = context
-				.getBean(JobLauncherCommandLineRunner.class);
+		JobLauncherApplicationRunner jobLauncherApplicationRunner = context
+				.getBean(TaskJobLauncherApplicationRunner.class);
 
-		Object names = ReflectionTestUtils.getField(jobLauncherCommandLineRunner,
+		Object names = ReflectionTestUtils.getField(jobLauncherApplicationRunner,
 				"jobNames");
 		assertThat(names).isEqualTo(jobNames);
 	}