1 | /* |
2 | * Copyright 2006-2013 the original author or authors. |
3 | * |
4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | * you may not use this file except in compliance with the License. |
6 | * You may obtain a copy of the License at |
7 | * |
8 | * http://www.apache.org/licenses/LICENSE-2.0 |
9 | * |
10 | * Unless required by applicable law or agreed to in writing, software |
11 | * distributed under the License is distributed on an "AS IS" BASIS, |
12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
13 | * See the License for the specific language governing permissions and |
14 | * limitations under the License. |
15 | */ |
16 | package org.springframework.batch.core.launch.support; |
17 | |
18 | import org.apache.commons.logging.Log; |
19 | import org.apache.commons.logging.LogFactory; |
20 | import org.springframework.batch.core.BatchStatus; |
21 | import org.springframework.batch.core.ExitStatus; |
22 | import org.springframework.batch.core.Job; |
23 | import org.springframework.batch.core.JobExecution; |
24 | import org.springframework.batch.core.JobInstance; |
25 | import org.springframework.batch.core.JobParameters; |
26 | import org.springframework.batch.core.JobParametersInvalidException; |
27 | import org.springframework.batch.core.StepExecution; |
28 | import org.springframework.batch.core.launch.JobLauncher; |
29 | import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; |
30 | import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; |
31 | import org.springframework.batch.core.repository.JobRepository; |
32 | import org.springframework.batch.core.repository.JobRestartException; |
33 | import org.springframework.beans.factory.InitializingBean; |
34 | import org.springframework.core.task.SyncTaskExecutor; |
35 | import org.springframework.core.task.TaskExecutor; |
36 | import org.springframework.core.task.TaskRejectedException; |
37 | import org.springframework.util.Assert; |
38 | |
39 | /** |
40 | * Simple implementation of the {@link JobLauncher} interface. The Spring Core |
41 | * {@link TaskExecutor} interface is used to launch a {@link Job}. This means |
42 | * that the type of executor set is very important. If a |
43 | * {@link SyncTaskExecutor} is used, then the job will be processed |
44 | * <strong>within the same thread that called the launcher.</strong> Care should |
45 | * be taken to ensure any users of this class understand fully whether or not |
46 | * the implementation of TaskExecutor used will start tasks synchronously or |
47 | * asynchronously. The default setting uses a synchronous task executor. |
48 | * |
49 | * There is only one required dependency of this Launcher, a |
50 | * {@link JobRepository}. The JobRepository is used to obtain a valid |
51 | * JobExecution. The Repository must be used because the provided {@link Job} |
52 | * could be a restart of an existing {@link JobInstance}, and only the |
53 | * Repository can reliably recreate it. |
54 | * |
55 | * @author Lucas Ward |
56 | * @Author Dave Syer |
57 | * @author Will Schipp |
58 | * |
59 | * @since 1.0 |
60 | * |
61 | * @see JobRepository |
62 | * @see TaskExecutor |
63 | */ |
64 | public class SimpleJobLauncher implements JobLauncher, InitializingBean { |
65 | |
66 | protected static final Log logger = LogFactory.getLog(SimpleJobLauncher.class); |
67 | |
68 | private JobRepository jobRepository; |
69 | |
70 | private TaskExecutor taskExecutor; |
71 | |
72 | /** |
73 | * Run the provided job with the given {@link JobParameters}. The |
74 | * {@link JobParameters} will be used to determine if this is an execution |
75 | * of an existing job instance, or if a new one should be created. |
76 | * |
77 | * @param job the job to be run. |
78 | * @param jobParameters the {@link JobParameters} for this particular |
79 | * execution. |
80 | * @return JobExecutionAlreadyRunningException if the JobInstance already |
81 | * exists and has an execution already running. |
82 | * @throws JobRestartException if the execution would be a re-start, but a |
83 | * re-start is either not allowed or not needed. |
84 | * @throws JobInstanceAlreadyCompleteException if this instance has already |
85 | * completed successfully |
86 | * @throws JobParametersInvalidException |
87 | */ |
88 | @Override |
89 | public JobExecution run(final Job job, final JobParameters jobParameters) |
90 | throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, |
91 | JobParametersInvalidException { |
92 | |
93 | Assert.notNull(job, "The Job must not be null."); |
94 | Assert.notNull(jobParameters, "The JobParameters must not be null."); |
95 | |
96 | final JobExecution jobExecution; |
97 | JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters); |
98 | if (lastExecution != null) { |
99 | if (!job.isRestartable()) { |
100 | throw new JobRestartException("JobInstance already exists and is not restartable"); |
101 | } |
102 | /* |
103 | * validate here if it has stepExecutions that are UNKNOWN |
104 | * retrieve the previous execution and check |
105 | */ |
106 | for (StepExecution execution : lastExecution.getStepExecutions()) { |
107 | if (execution.getStatus() == BatchStatus.UNKNOWN) { |
108 | //throw |
109 | throw new JobRestartException("Step [" + execution.getStepName() + "] is of status UNKNOWN"); |
110 | }//end if |
111 | }//end for |
112 | } |
113 | |
114 | // Check the validity of the parameters before doing creating anything |
115 | // in the repository... |
116 | job.getJobParametersValidator().validate(jobParameters); |
117 | |
118 | /* |
119 | * There is a very small probability that a non-restartable job can be |
120 | * restarted, but only if another process or thread manages to launch |
121 | * <i>and</i> fail a job execution for this instance between the last |
122 | * assertion and the next method returning successfully. |
123 | */ |
124 | jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters); |
125 | |
126 | try { |
127 | taskExecutor.execute(new Runnable() { |
128 | |
129 | @Override |
130 | public void run() { |
131 | try { |
132 | logger.info("Job: [" + job + "] launched with the following parameters: [" + jobParameters |
133 | + "]"); |
134 | job.execute(jobExecution); |
135 | logger.info("Job: [" + job + "] completed with the following parameters: [" + jobParameters |
136 | + "] and the following status: [" + jobExecution.getStatus() + "]"); |
137 | } |
138 | catch (Throwable t) { |
139 | logger.info("Job: [" + job |
140 | + "] failed unexpectedly and fatally with the following parameters: [" + jobParameters |
141 | + "]", t); |
142 | rethrow(t); |
143 | } |
144 | } |
145 | |
146 | private void rethrow(Throwable t) { |
147 | if (t instanceof RuntimeException) { |
148 | throw (RuntimeException) t; |
149 | } |
150 | else if (t instanceof Error) { |
151 | throw (Error) t; |
152 | } |
153 | throw new IllegalStateException(t); |
154 | } |
155 | }); |
156 | } |
157 | catch (TaskRejectedException e) { |
158 | jobExecution.upgradeStatus(BatchStatus.FAILED); |
159 | if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) { |
160 | jobExecution.setExitStatus(ExitStatus.FAILED.addExitDescription(e)); |
161 | } |
162 | jobRepository.update(jobExecution); |
163 | } |
164 | |
165 | return jobExecution; |
166 | } |
167 | |
168 | /** |
169 | * Set the JobRepsitory. |
170 | * |
171 | * @param jobRepository |
172 | */ |
173 | public void setJobRepository(JobRepository jobRepository) { |
174 | this.jobRepository = jobRepository; |
175 | } |
176 | |
177 | /** |
178 | * Set the TaskExecutor. (Optional) |
179 | * |
180 | * @param taskExecutor |
181 | */ |
182 | public void setTaskExecutor(TaskExecutor taskExecutor) { |
183 | this.taskExecutor = taskExecutor; |
184 | } |
185 | |
186 | /** |
187 | * Ensure the required dependencies of a {@link JobRepository} have been |
188 | * set. |
189 | */ |
190 | @Override |
191 | public void afterPropertiesSet() throws Exception { |
192 | Assert.state(jobRepository != null, "A JobRepository has not been set."); |
193 | if (taskExecutor == null) { |
194 | logger.info("No TaskExecutor has been set, defaulting to synchronous executor."); |
195 | taskExecutor = new SyncTaskExecutor(); |
196 | } |
197 | } |
198 | |
199 | } |