From 2cba0f223aca726b21f27a4031d33d5f49f7e551 Mon Sep 17 00:00:00 2001 From: Glenn Renfro Date: Tue, 14 Jul 2020 14:12:59 -0400 Subject: [PATCH] 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 --- ...bLauncherApplicationRunnerFactoryBean.java | 107 ++++++++ .../TaskJobLauncherAutoConfiguration.java | 20 +- ...bLauncherCommandLineRunnerFactoryBean.java | 4 +- .../TaskJobLauncherApplicationRunner.java | 234 ++++++++++++++++++ .../TaskJobLauncherCommandLineRunner.java | 3 +- ...TaskJobLauncherAutoConfigurationTests.java | 22 +- 6 files changed, 377 insertions(+), 13 deletions(-) create mode 100644 spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherApplicationRunnerFactoryBean.java create mode 100644 spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunner.java diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherApplicationRunnerFactoryBean.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherApplicationRunnerFactoryBean.java new file mode 100644 index 00000000..e689a0b6 --- /dev/null +++ b/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 { + + private JobLauncher jobLauncher; + + private JobExplorer jobExplorer; + + private List jobs; + + private String jobNames; + + private JobRegistry jobRegistry; + + private Integer order = 0; + + private TaskBatchProperties taskBatchProperties; + + private JobRepository jobRepository; + + public TaskJobLauncherApplicationRunnerFactoryBean(JobLauncher jobLauncher, + JobExplorer jobExplorer, List 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; + } + +} diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfiguration.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfiguration.java index 4df9f23b..9e98ca9c 100644 --- a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfiguration.java +++ b/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 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 jobs, + JobRegistry jobRegistry, JobRepository jobRepository, + BatchProperties batchProperties) { + TaskJobLauncherApplicationRunnerFactoryBean taskJobLauncherApplicationRunnerFactoryBean; + taskJobLauncherApplicationRunnerFactoryBean = new TaskJobLauncherApplicationRunnerFactoryBean( + jobLauncher, jobExplorer, jobs, this.properties, jobRegistry, + jobRepository, batchProperties); + + return taskJobLauncherApplicationRunnerFactoryBean; + } + } diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherCommandLineRunnerFactoryBean.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherCommandLineRunnerFactoryBean.java index abd314a8..54f3bf64 100644 --- a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherCommandLineRunnerFactoryBean.java +++ b/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 { diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunner.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherApplicationRunner.java new file mode 100644 index 00000000..ce36b845 --- /dev/null +++ b/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 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 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 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 parameterMap = parameters.getParameters(); + HashMap copy = new HashMap<>(parameterMap); + + for (Map.Entry 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 merged = new HashMap<>(); + merged.putAll(parameters.getParameters()); + merged.putAll(additionals.getParameters()); + return new JobParameters(merged); + } + +} diff --git a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherCommandLineRunner.java b/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherCommandLineRunner.java index 18944163..b8166d6f 100644 --- a/spring-cloud-task-batch/src/main/java/org/springframework/cloud/task/batch/handler/TaskJobLauncherCommandLineRunner.java +++ b/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 { diff --git a/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfigurationTests.java b/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfigurationTests.java index f95f561f..d00fc345 100644 --- a/spring-cloud-task-batch/src/test/java/org/springframework/cloud/task/batch/configuration/TaskJobLauncherAutoConfigurationTests.java +++ b/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); }