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 | |
17 | package org.springframework.batch.core.job; |
18 | |
19 | import java.util.Collection; |
20 | import java.util.Date; |
21 | |
22 | import org.apache.commons.logging.Log; |
23 | import org.apache.commons.logging.LogFactory; |
24 | import org.springframework.batch.core.BatchStatus; |
25 | import org.springframework.batch.core.ExitStatus; |
26 | import org.springframework.batch.core.Job; |
27 | import org.springframework.batch.core.JobExecution; |
28 | import org.springframework.batch.core.JobExecutionException; |
29 | import org.springframework.batch.core.JobExecutionListener; |
30 | import org.springframework.batch.core.JobInterruptedException; |
31 | import org.springframework.batch.core.JobParametersIncrementer; |
32 | import org.springframework.batch.core.JobParametersValidator; |
33 | import org.springframework.batch.core.StartLimitExceededException; |
34 | import org.springframework.batch.core.Step; |
35 | import org.springframework.batch.core.StepExecution; |
36 | import org.springframework.batch.core.launch.NoSuchJobException; |
37 | import org.springframework.batch.core.launch.support.ExitCodeMapper; |
38 | import org.springframework.batch.core.listener.CompositeJobExecutionListener; |
39 | import org.springframework.batch.core.repository.JobRepository; |
40 | import org.springframework.batch.core.repository.JobRestartException; |
41 | import org.springframework.batch.core.step.StepLocator; |
42 | import org.springframework.batch.repeat.RepeatException; |
43 | import org.springframework.beans.factory.BeanNameAware; |
44 | import org.springframework.beans.factory.InitializingBean; |
45 | import org.springframework.util.Assert; |
46 | import org.springframework.util.ClassUtils; |
47 | |
48 | /** |
49 | * Abstract implementation of the {@link Job} interface. Common dependencies |
50 | * such as a {@link JobRepository}, {@link JobExecutionListener}s, and various |
51 | * configuration parameters are set here. Therefore, common error handling and |
52 | * listener calling activities are abstracted away from implementations. |
53 | * |
54 | * @author Lucas Ward |
55 | * @author Dave Syer |
56 | */ |
57 | public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, |
58 | InitializingBean { |
59 | |
60 | protected static final Log logger = LogFactory.getLog(AbstractJob.class); |
61 | |
62 | private String name; |
63 | |
64 | private boolean restartable = true; |
65 | |
66 | private JobRepository jobRepository; |
67 | |
68 | private CompositeJobExecutionListener listener = new CompositeJobExecutionListener(); |
69 | |
70 | private JobParametersIncrementer jobParametersIncrementer; |
71 | |
72 | private JobParametersValidator jobParametersValidator = new DefaultJobParametersValidator(); |
73 | |
74 | private StepHandler stepHandler; |
75 | |
76 | /** |
77 | * Default constructor. |
78 | */ |
79 | public AbstractJob() { |
80 | super(); |
81 | } |
82 | |
83 | /** |
84 | * Convenience constructor to immediately add name (which is mandatory but |
85 | * not final). |
86 | * |
87 | * @param name |
88 | */ |
89 | public AbstractJob(String name) { |
90 | super(); |
91 | this.name = name; |
92 | } |
93 | |
94 | /** |
95 | * A validator for job parameters. Defaults to a vanilla |
96 | * {@link DefaultJobParametersValidator}. |
97 | * |
98 | * @param jobParametersValidator |
99 | * a validator instance |
100 | */ |
101 | public void setJobParametersValidator( |
102 | JobParametersValidator jobParametersValidator) { |
103 | this.jobParametersValidator = jobParametersValidator; |
104 | } |
105 | |
106 | /** |
107 | * Assert mandatory properties: {@link JobRepository}. |
108 | * |
109 | * @see InitializingBean#afterPropertiesSet() |
110 | */ |
111 | @Override |
112 | public void afterPropertiesSet() throws Exception { |
113 | Assert.notNull(jobRepository, "JobRepository must be set"); |
114 | } |
115 | |
116 | /** |
117 | * Set the name property if it is not already set. Because of the order of |
118 | * the callbacks in a Spring container the name property will be set first |
119 | * if it is present. Care is needed with bean definition inheritance - if a |
120 | * parent bean has a name, then its children need an explicit name as well, |
121 | * otherwise they will not be unique. |
122 | * |
123 | * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) |
124 | */ |
125 | @Override |
126 | public void setBeanName(String name) { |
127 | if (this.name == null) { |
128 | this.name = name; |
129 | } |
130 | } |
131 | |
132 | /** |
133 | * Set the name property. Always overrides the default value if this object |
134 | * is a Spring bean. |
135 | * |
136 | * @see #setBeanName(java.lang.String) |
137 | */ |
138 | public void setName(String name) { |
139 | this.name = name; |
140 | } |
141 | |
142 | /* |
143 | * (non-Javadoc) |
144 | * |
145 | * @see org.springframework.batch.core.domain.IJob#getName() |
146 | */ |
147 | @Override |
148 | public String getName() { |
149 | return name; |
150 | } |
151 | |
152 | /** |
153 | * Retrieve the step with the given name. If there is no Step with the given |
154 | * name, then return null. |
155 | * |
156 | * @param stepName |
157 | * @return the Step |
158 | */ |
159 | @Override |
160 | public abstract Step getStep(String stepName); |
161 | |
162 | /** |
163 | * Retrieve the step names. |
164 | * |
165 | * @return the step names |
166 | */ |
167 | @Override |
168 | public abstract Collection<String> getStepNames(); |
169 | |
170 | @Override |
171 | public JobParametersValidator getJobParametersValidator() { |
172 | return jobParametersValidator; |
173 | } |
174 | |
175 | /** |
176 | * Boolean flag to prevent categorically a job from restarting, even if it |
177 | * has failed previously. |
178 | * |
179 | * @param restartable |
180 | * the value of the flag to set (default true) |
181 | */ |
182 | public void setRestartable(boolean restartable) { |
183 | this.restartable = restartable; |
184 | } |
185 | |
186 | /** |
187 | * @see Job#isRestartable() |
188 | */ |
189 | @Override |
190 | public boolean isRestartable() { |
191 | return restartable; |
192 | } |
193 | |
194 | /** |
195 | * Public setter for the {@link JobParametersIncrementer}. |
196 | * |
197 | * @param jobParametersIncrementer |
198 | * the {@link JobParametersIncrementer} to set |
199 | */ |
200 | public void setJobParametersIncrementer( |
201 | JobParametersIncrementer jobParametersIncrementer) { |
202 | this.jobParametersIncrementer = jobParametersIncrementer; |
203 | } |
204 | |
205 | /* |
206 | * (non-Javadoc) |
207 | * |
208 | * @see org.springframework.batch.core.Job#getJobParametersIncrementer() |
209 | */ |
210 | @Override |
211 | public JobParametersIncrementer getJobParametersIncrementer() { |
212 | return this.jobParametersIncrementer; |
213 | } |
214 | |
215 | /** |
216 | * Public setter for injecting {@link JobExecutionListener}s. They will all |
217 | * be given the listener callbacks at the appropriate point in the job. |
218 | * |
219 | * @param listeners |
220 | * the listeners to set. |
221 | */ |
222 | public void setJobExecutionListeners(JobExecutionListener[] listeners) { |
223 | for (int i = 0; i < listeners.length; i++) { |
224 | this.listener.register(listeners[i]); |
225 | } |
226 | } |
227 | |
228 | /** |
229 | * Register a single listener for the {@link JobExecutionListener} |
230 | * callbacks. |
231 | * |
232 | * @param listener |
233 | * a {@link JobExecutionListener} |
234 | */ |
235 | public void registerJobExecutionListener(JobExecutionListener listener) { |
236 | this.listener.register(listener); |
237 | } |
238 | |
239 | /** |
240 | * Public setter for the {@link JobRepository} that is needed to manage the |
241 | * state of the batch meta domain (jobs, steps, executions) during the life |
242 | * of a job. |
243 | * |
244 | * @param jobRepository |
245 | */ |
246 | public void setJobRepository(JobRepository jobRepository) { |
247 | this.jobRepository = jobRepository; |
248 | stepHandler = new SimpleStepHandler(jobRepository); |
249 | } |
250 | |
251 | /** |
252 | * Convenience method for subclasses to access the job repository. |
253 | * |
254 | * @return the jobRepository |
255 | */ |
256 | protected JobRepository getJobRepository() { |
257 | return jobRepository; |
258 | } |
259 | |
260 | /** |
261 | * Extension point for subclasses allowing them to concentrate on processing |
262 | * logic and ignore listeners and repository calls. Implementations usually |
263 | * are concerned with the ordering of steps, and delegate actual step |
264 | * processing to {@link #handleStep(Step, JobExecution)}. |
265 | * |
266 | * @param execution |
267 | * the current {@link JobExecution} |
268 | * |
269 | * @throws JobExecutionException |
270 | * to signal a fatal batch framework error (not a business or |
271 | * validation exception) |
272 | */ |
273 | abstract protected void doExecute(JobExecution execution) |
274 | throws JobExecutionException; |
275 | |
276 | /** |
277 | * Run the specified job, handling all listener and repository calls, and |
278 | * delegating the actual processing to {@link #doExecute(JobExecution)}. |
279 | * |
280 | * @see Job#execute(JobExecution) |
281 | * @throws StartLimitExceededException |
282 | * if start limit of one of the steps was exceeded |
283 | */ |
284 | @Override |
285 | public final void execute(JobExecution execution) { |
286 | |
287 | logger.debug("Job execution starting: " + execution); |
288 | |
289 | try { |
290 | |
291 | jobParametersValidator.validate(execution.getJobParameters()); |
292 | |
293 | if (execution.getStatus() != BatchStatus.STOPPING) { |
294 | |
295 | execution.setStartTime(new Date()); |
296 | updateStatus(execution, BatchStatus.STARTED); |
297 | |
298 | listener.beforeJob(execution); |
299 | |
300 | try { |
301 | doExecute(execution); |
302 | logger.debug("Job execution complete: " + execution); |
303 | } catch (RepeatException e) { |
304 | throw e.getCause(); |
305 | } |
306 | } else { |
307 | |
308 | // The job was already stopped before we even got this far. Deal |
309 | // with it in the same way as any other interruption. |
310 | execution.setStatus(BatchStatus.STOPPED); |
311 | execution.setExitStatus(ExitStatus.COMPLETED); |
312 | logger.debug("Job execution was stopped: " + execution); |
313 | |
314 | } |
315 | |
316 | } catch (JobInterruptedException e) { |
317 | logger.info("Encountered interruption executing job: " |
318 | + e.getMessage()); |
319 | if (logger.isDebugEnabled()) { |
320 | logger.debug("Full exception", e); |
321 | } |
322 | execution.setExitStatus(getDefaultExitStatusForFailure(e)); |
323 | execution.setStatus(BatchStatus.max(BatchStatus.STOPPED, e.getStatus())); |
324 | execution.addFailureException(e); |
325 | } catch (Throwable t) { |
326 | logger.error("Encountered fatal error executing job", t); |
327 | execution.setExitStatus(getDefaultExitStatusForFailure(t)); |
328 | execution.setStatus(BatchStatus.FAILED); |
329 | execution.addFailureException(t); |
330 | } finally { |
331 | |
332 | if (execution.getStatus().isLessThanOrEqualTo(BatchStatus.STOPPED) |
333 | && execution.getStepExecutions().isEmpty()) { |
334 | ExitStatus exitStatus = execution.getExitStatus(); |
335 | execution |
336 | .setExitStatus(exitStatus.and(ExitStatus.NOOP |
337 | .addExitDescription("All steps already completed or no steps configured for this job."))); |
338 | } |
339 | |
340 | execution.setEndTime(new Date()); |
341 | |
342 | try { |
343 | listener.afterJob(execution); |
344 | } catch (Exception e) { |
345 | logger.error("Exception encountered in afterStep callback", e); |
346 | } |
347 | |
348 | jobRepository.update(execution); |
349 | |
350 | } |
351 | |
352 | } |
353 | |
354 | /** |
355 | * Convenience method for subclasses to delegate the handling of a specific |
356 | * step in the context of the current {@link JobExecution}. Clients of this |
357 | * method do not need access to the {@link JobRepository}, nor do they need |
358 | * to worry about populating the execution context on a restart, nor |
359 | * detecting the interrupted state (in job or step execution). |
360 | * |
361 | * @param step |
362 | * the {@link Step} to execute |
363 | * @param execution |
364 | * the current {@link JobExecution} |
365 | * @return the {@link StepExecution} corresponding to this step |
366 | * |
367 | * @throws JobInterruptedException |
368 | * if the {@link JobExecution} has been interrupted, and in |
369 | * particular if {@link BatchStatus#ABANDONED} or |
370 | * {@link BatchStatus#STOPPING} is detected |
371 | * @throws StartLimitExceededException |
372 | * if the start limit has been exceeded for this step |
373 | * @throws JobRestartException |
374 | * if the job is in an inconsistent state from an earlier |
375 | * failure |
376 | */ |
377 | protected final StepExecution handleStep(Step step, JobExecution execution) |
378 | throws JobInterruptedException, JobRestartException, |
379 | StartLimitExceededException { |
380 | return stepHandler.handleStep(step, execution); |
381 | |
382 | } |
383 | |
384 | /** |
385 | * Default mapping from throwable to {@link ExitStatus}. |
386 | * |
387 | * @param ex |
388 | * the cause of the failure |
389 | * @return an {@link ExitStatus} |
390 | */ |
391 | private ExitStatus getDefaultExitStatusForFailure(Throwable ex) { |
392 | ExitStatus exitStatus; |
393 | if (ex instanceof JobInterruptedException |
394 | || ex.getCause() instanceof JobInterruptedException) { |
395 | exitStatus = ExitStatus.STOPPED |
396 | .addExitDescription(JobInterruptedException.class.getName()); |
397 | } else if (ex instanceof NoSuchJobException |
398 | || ex.getCause() instanceof NoSuchJobException) { |
399 | exitStatus = new ExitStatus(ExitCodeMapper.NO_SUCH_JOB, ex |
400 | .getClass().getName()); |
401 | } else { |
402 | exitStatus = ExitStatus.FAILED.addExitDescription(ex); |
403 | } |
404 | |
405 | return exitStatus; |
406 | } |
407 | |
408 | private void updateStatus(JobExecution jobExecution, BatchStatus status) { |
409 | jobExecution.setStatus(status); |
410 | jobRepository.update(jobExecution); |
411 | } |
412 | |
413 | @Override |
414 | public String toString() { |
415 | return ClassUtils.getShortName(getClass()) + ": [name=" + name + "]"; |
416 | } |
417 | |
418 | } |