Skip to content

Commit 7d35e0d

Browse files
cppwfscppwfs
committed
Updated TaskJobLauncherCommandLineRunner to match what is in Spring Boot 2.1.x
The changes to the pom files will need to be removed when Spring Cloud is updated to the latest release. resolves spring-cloud#466
1 parent 6c9bce9 commit 7d35e0d

File tree

7 files changed

+234
-16
lines changed

7 files changed

+234
-16
lines changed

‎pom.xml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -142,6 +142,7 @@
142142
<commons-logging.version>1.1</commons-logging.version>
143143
<java-ee-api.version>8.0</java-ee-api.version>
144144
<junit.version>5.3.1</junit.version>
145+
<spring-boot.version>2.1.0.BUILD-SNAPSHOT</spring-boot.version>
145146

146147
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
147148
<sonar.jacoco.reportPath>${project.build.directory}/coverage-reports/jacoco-ut.exec</sonar.jacoco.reportPath>

‎spring-cloud-task-batch/pom.xml

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@
6969
<groupId>org.springframework.boot</groupId>
7070
<artifactId>spring-boot-configuration-processor</artifactId>
7171
<optional>true</optional>
72+
<version>${spring-boot.version}</version>
7273
</dependency>
7374
<dependency>
7475
<groupId>org.assertj</groupId>
@@ -78,6 +79,32 @@
7879
<dependency>
7980
<groupId>org.springframework.boot</groupId>
8081
<artifactId>spring-boot-test</artifactId>
82+
<version>${spring-boot.version}</version>
83+
</dependency>
84+
<dependency>
85+
<groupId>org.springframework.boot</groupId>
86+
<artifactId>spring-boot</artifactId>
87+
<version>${spring-boot.version}</version>
88+
</dependency>
89+
<dependency>
90+
<groupId>org.springframework.boot</groupId>
91+
<artifactId>spring-boot-autoconfigure</artifactId>
92+
<version>${spring-boot.version}</version>
93+
</dependency>
94+
<dependency>
95+
<groupId>org.springframework.boot</groupId>
96+
<artifactId>spring-boot-starter</artifactId>
97+
<version>${spring-boot.version}</version>
98+
</dependency>
99+
<dependency>
100+
<groupId>org.springframework.boot</groupId>
101+
<artifactId>spring-boot-starter-logging</artifactId>
102+
<version>${spring-boot.version}</version>
103+
</dependency>
104+
<dependency>
105+
<groupId>org.springframework.boot</groupId>
106+
<artifactId>spring-boot-starter-validation</artifactId>
107+
<version>${spring-boot.version}</version>
81108
</dependency>
82109
<dependency>
83110
<groupId>org.junit.jupiter</groupId>

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

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,19 @@
1818

1919
import java.util.List;
2020

21+
import javax.sql.DataSource;
22+
2123
import org.springframework.batch.core.Job;
2224
import org.springframework.batch.core.configuration.JobRegistry;
2325
import org.springframework.batch.core.explore.JobExplorer;
2426
import org.springframework.batch.core.launch.JobLauncher;
27+
import org.springframework.batch.core.repository.JobRepository;
28+
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
29+
import org.springframework.batch.core.repository.support.SimpleJobRepository;
2530
import org.springframework.beans.factory.annotation.Autowired;
2631
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
2732
import org.springframework.boot.autoconfigure.batch.BatchAutoConfiguration;
33+
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
2834
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
2935
import org.springframework.boot.context.properties.EnableConfigurationProperties;
3036
import org.springframework.context.annotation.Bean;
@@ -45,15 +51,25 @@ public class TaskJobLauncherAutoConfiguration {
4551
@Autowired
4652
private TaskBatchProperties properties;
4753

54+
55+
@Bean
56+
@ConditionalOnMissingBean(JobRepository.class)
57+
public JobRepository jobRepository(DataSource dataSource) throws Exception{
58+
JobRepositoryFactoryBean factoryBean = new JobRepositoryFactoryBean();
59+
factoryBean.setDataSource(dataSource);
60+
return factoryBean.getObject();
61+
}
62+
4863
@Bean
4964
public TaskJobLauncherCommandLineRunnerFactoryBean jobLauncherCommandLineRunner(JobLauncher jobLauncher,
50-
JobExplorer jobExplorer, List<Job> jobs, JobRegistry jobRegistry) {
65+
JobExplorer jobExplorer, List<Job> jobs, JobRegistry jobRegistry, JobRepository jobRepository) {
5166
TaskJobLauncherCommandLineRunnerFactoryBean taskJobLauncherCommandLineRunnerFactoryBean =
5267
new TaskJobLauncherCommandLineRunnerFactoryBean(jobLauncher,
5368
jobExplorer,
5469
jobs,
5570
this.properties,
56-
jobRegistry);
71+
jobRegistry,
72+
jobRepository);
5773

5874
return taskJobLauncherCommandLineRunnerFactoryBean;
5975
}

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

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.springframework.batch.core.configuration.JobRegistry;
2323
import org.springframework.batch.core.explore.JobExplorer;
2424
import org.springframework.batch.core.launch.JobLauncher;
25+
import org.springframework.batch.core.repository.JobRepository;
2526
import org.springframework.beans.factory.FactoryBean;
2627
import org.springframework.cloud.task.batch.handler.TaskJobLauncherCommandLineRunner;
2728
import org.springframework.util.Assert;
@@ -48,9 +49,11 @@ public class TaskJobLauncherCommandLineRunnerFactoryBean implements FactoryBean<
4849

4950
private TaskBatchProperties taskBatchProperties;
5051

52+
private JobRepository jobRepository;
53+
5154
public TaskJobLauncherCommandLineRunnerFactoryBean(JobLauncher jobLauncher,
5255
JobExplorer jobExplorer, List<Job> jobs, TaskBatchProperties taskBatchProperties,
53-
JobRegistry jobRegistry) {
56+
JobRegistry jobRegistry, JobRepository jobRepository) {
5457
Assert.notNull(taskBatchProperties, "properties must not be null");
5558
this.jobLauncher = jobLauncher;
5659
this.jobExplorer = jobExplorer;
@@ -60,6 +63,7 @@ public TaskJobLauncherCommandLineRunnerFactoryBean(JobLauncher jobLauncher,
6063
this.jobRegistry = jobRegistry;
6164
this.taskBatchProperties = taskBatchProperties;
6265
this.order = taskBatchProperties.getCommandLineRunnerOrder();
66+
this.jobRepository = jobRepository;
6367
}
6468

6569
public void setOrder(int order) {
@@ -69,7 +73,7 @@ public void setOrder(int order) {
6973
@Override
7074
public TaskJobLauncherCommandLineRunner getObject() {
7175
TaskJobLauncherCommandLineRunner taskJobLauncherCommandLineRunner =
72-
new TaskJobLauncherCommandLineRunner(this.jobLauncher, this.jobExplorer, this.taskBatchProperties);
76+
new TaskJobLauncherCommandLineRunner(this.jobLauncher, this.jobExplorer, this.jobRepository, this.taskBatchProperties);
7377
taskJobLauncherCommandLineRunner.setJobs(this.jobs);
7478
if(StringUtils.hasText(this.jobNames)) {
7579
taskJobLauncherCommandLineRunner.setJobNames(this.jobNames);

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

Lines changed: 72 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,9 @@
2020
import java.util.Arrays;
2121
import java.util.Collections;
2222
import java.util.Date;
23+
import java.util.HashMap;
2324
import java.util.List;
25+
import java.util.Map;
2426

2527
import org.apache.commons.logging.Log;
2628
import org.apache.commons.logging.LogFactory;
@@ -29,13 +31,16 @@
2931
import org.springframework.batch.core.Job;
3032
import org.springframework.batch.core.JobExecution;
3133
import org.springframework.batch.core.JobExecutionException;
34+
import org.springframework.batch.core.JobParameter;
3235
import org.springframework.batch.core.JobParameters;
3336
import org.springframework.batch.core.JobParametersBuilder;
37+
import org.springframework.batch.core.JobParametersIncrementer;
3438
import org.springframework.batch.core.JobParametersInvalidException;
3539
import org.springframework.batch.core.explore.JobExplorer;
3640
import org.springframework.batch.core.launch.JobLauncher;
3741
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
3842
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
43+
import org.springframework.batch.core.repository.JobRepository;
3944
import org.springframework.batch.core.repository.JobRestartException;
4045
import org.springframework.batch.repeat.RepeatCallback;
4146
import org.springframework.batch.repeat.RepeatContext;
@@ -71,6 +76,8 @@ public class TaskJobLauncherCommandLineRunner extends JobLauncherCommandLineRunn
7176

7277
private JobExplorer taskJobExplorer;
7378

79+
private JobRepository taskJobRepository;
80+
7481
private static final Log logger = LogFactory
7582
.getLog(TaskJobLauncherCommandLineRunner.class);
7683

@@ -80,11 +87,20 @@ public class TaskJobLauncherCommandLineRunner extends JobLauncherCommandLineRunn
8087

8188
private TaskBatchProperties taskBatchProperties;
8289

83-
public TaskJobLauncherCommandLineRunner(JobLauncher jobLauncher,
84-
JobExplorer jobExplorer, TaskBatchProperties taskBatchProperties) {
85-
super(jobLauncher, jobExplorer);
90+
/**
91+
* Create a new {@link TaskJobLauncherCommandLineRunner}.
92+
* @param jobLauncher to launch jobs
93+
* @param jobExplorer to check the job repository for previous executions
94+
* @param jobRepository to check if a job instance exists with the given parameters
95+
* when running a job
96+
* @param taskBatchProperties the properties used to configure the taskBatchProperties.
97+
*/
98+
public TaskJobLauncherCommandLineRunner(JobLauncher jobLauncher, JobExplorer jobExplorer,
99+
JobRepository jobRepository, TaskBatchProperties taskBatchProperties) {
100+
super(jobLauncher, jobExplorer, jobRepository);
86101
this.taskJobLauncher = jobLauncher;
87102
this.taskJobExplorer = jobExplorer;
103+
this.taskJobRepository = jobRepository;
88104
this.taskBatchProperties = taskBatchProperties;
89105
}
90106

@@ -103,9 +119,39 @@ public void run(String... args) throws JobExecutionException {
103119
protected void execute(Job job, JobParameters jobParameters)
104120
throws JobExecutionAlreadyRunningException, JobRestartException,
105121
JobInstanceAlreadyCompleteException, JobParametersInvalidException {
106-
JobParameters nextParameters = new JobParametersBuilder(jobParameters,
107-
this.taskJobExplorer).getNextJobParameters(job).toJobParameters();
108-
JobExecution execution = this.taskJobLauncher.run(job, nextParameters);
122+
String jobName = job.getName();
123+
JobParameters parameters = jobParameters;
124+
boolean jobInstanceExists = this.taskJobRepository.isJobInstanceExists(jobName,
125+
parameters);
126+
if (jobInstanceExists) {
127+
JobExecution lastJobExecution = this.taskJobRepository
128+
.getLastJobExecution(jobName, jobParameters);
129+
if (lastJobExecution != null && isStoppedOrFailed(lastJobExecution)
130+
&& job.isRestartable()) {
131+
// Retry a failed or stopped execution with previous parameters
132+
JobParameters previousParameters = lastJobExecution.getJobParameters();
133+
/*
134+
* remove Non-identifying parameters from the previous execution's
135+
* parameters since there is no way to remove them programmatically. If
136+
* they are required (or need to be modified) on a restart, they need to
137+
* be (re)specified.
138+
*/
139+
JobParameters previousIdentifyingParameters = removeNonIdentifying(
140+
previousParameters);
141+
// merge additional parameters with previous ones (overriding those with
142+
// the same key)
143+
parameters = merge(previousIdentifyingParameters, jobParameters);
144+
}
145+
}
146+
else {
147+
JobParametersIncrementer incrementer = job.getJobParametersIncrementer();
148+
if (incrementer != null) {
149+
JobParameters nextParameters = new JobParametersBuilder(jobParameters,
150+
this.taskJobExplorer).getNextJobParameters(job).toJobParameters();
151+
parameters = merge(nextParameters, jobParameters);
152+
}
153+
}
154+
JobExecution execution = this.taskJobLauncher.run(job, parameters);
109155
if (this.taskApplicationEventPublisher != null) {
110156
this.taskApplicationEventPublisher.publishEvent(new JobExecutionEvent(execution));
111157
}
@@ -164,5 +210,24 @@ public void throwJobFailedException(List<JobExecution> failedJobExecutions) {
164210
throw new TaskException(message);
165211

166212
}
167-
213+
private JobParameters removeNonIdentifying(JobParameters parameters) {
214+
Map<String, JobParameter> parameterMap = parameters.getParameters();
215+
HashMap<String, JobParameter> copy = new HashMap<>(parameterMap);
216+
for (Map.Entry<String, JobParameter> parameter : copy.entrySet()) {
217+
if (!parameter.getValue().isIdentifying()) {
218+
parameterMap.remove(parameter.getKey());
219+
}
220+
}
221+
return new JobParameters(parameterMap);
222+
}
223+
private boolean isStoppedOrFailed(JobExecution execution) {
224+
BatchStatus status = execution.getStatus();
225+
return (status == BatchStatus.STOPPED || status == BatchStatus.FAILED);
226+
}
227+
private JobParameters merge(JobParameters parameters, JobParameters additionals) {
228+
Map<String, JobParameter> merged = new HashMap<>();
229+
merged.putAll(parameters.getParameters());
230+
merged.putAll(additionals.getParameters());
231+
return new JobParameters(merged);
232+
}
168233
}

0 commit comments

Comments
 (0)