1 | /* |
2 | * Copyright 2006-2007 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.Date; |
20 | import java.util.Iterator; |
21 | import java.util.List; |
22 | |
23 | import org.springframework.batch.core.BatchStatus; |
24 | import org.springframework.batch.core.JobExecution; |
25 | import org.springframework.batch.core.JobExecutionException; |
26 | import org.springframework.batch.core.JobInstance; |
27 | import org.springframework.batch.core.JobInterruptedException; |
28 | import org.springframework.batch.core.StartLimitExceededException; |
29 | import org.springframework.batch.core.Step; |
30 | import org.springframework.batch.core.StepExecution; |
31 | import org.springframework.batch.core.UnexpectedJobExecutionException; |
32 | import org.springframework.batch.item.ExecutionContext; |
33 | import org.springframework.batch.repeat.ExitStatus; |
34 | |
35 | /** |
36 | * Simple implementation of (@link Job} interface providing the ability to run a |
37 | * {@link JobExecution}. Sequentially executes a job by iterating through its |
38 | * list of steps. |
39 | * |
40 | * @author Lucas Ward |
41 | * @author Dave Syer |
42 | */ |
43 | public class SimpleJob extends AbstractJob { |
44 | |
45 | /** |
46 | * Run the specified job by looping through the steps and delegating to the |
47 | * {@link Step}. |
48 | * |
49 | * @see org.springframework.batch.core.Job#execute(org.springframework.batch.core.JobExecution) |
50 | * @throws StartLimitExceededException if start limit of one of the steps |
51 | * was exceeded |
52 | */ |
53 | public void execute(JobExecution execution) throws JobExecutionException { |
54 | |
55 | JobInstance jobInstance = execution.getJobInstance(); |
56 | |
57 | StepExecution currentStepExecution = null; |
58 | int startedCount = 0; |
59 | List steps = getSteps(); |
60 | |
61 | // holder for potential job failure to be passed to onError listener |
62 | Throwable jobFailure = null; |
63 | |
64 | try { |
65 | |
66 | // The job was already stopped before we even got this far. Deal |
67 | // with it in the same way as any other interruption. |
68 | if (execution.getStatus() == BatchStatus.STOPPING) { |
69 | throw new JobInterruptedException("JobExecution already stopped before being executed."); |
70 | } |
71 | |
72 | execution.setStartTime(new Date()); |
73 | updateStatus(execution, BatchStatus.STARTING); |
74 | |
75 | getCompositeListener().beforeJob(execution); |
76 | |
77 | for (Iterator i = steps.iterator(); i.hasNext();) { |
78 | |
79 | if (execution.getStatus() == BatchStatus.STOPPING) { |
80 | throw new JobInterruptedException("JobExecution interrupted."); |
81 | } |
82 | |
83 | Step step = (Step) i.next(); |
84 | |
85 | if (shouldStart(jobInstance, step)) { |
86 | |
87 | startedCount++; |
88 | updateStatus(execution, BatchStatus.STARTED); |
89 | currentStepExecution = execution.createStepExecution(step); |
90 | |
91 | StepExecution lastStepExecution = getJobRepository().getLastStepExecution(jobInstance, step); |
92 | |
93 | boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals( |
94 | BatchStatus.COMPLETED)) ? true : false; |
95 | |
96 | if (isRestart) { |
97 | currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext()); |
98 | } |
99 | else { |
100 | currentStepExecution.setExecutionContext(new ExecutionContext()); |
101 | } |
102 | |
103 | step.execute(currentStepExecution); |
104 | |
105 | } |
106 | } |
107 | |
108 | // Need to check again for stopped job |
109 | if (execution.getStatus() == BatchStatus.STOPPING) { |
110 | throw new JobInterruptedException("JobExecution interrupted."); |
111 | } |
112 | |
113 | updateStatus(execution, BatchStatus.COMPLETED); |
114 | |
115 | } |
116 | catch (JobInterruptedException e) { |
117 | execution.setStatus(BatchStatus.STOPPED); |
118 | rethrow(e); |
119 | } |
120 | catch (Throwable t) { |
121 | execution.setStatus(BatchStatus.FAILED); |
122 | jobFailure = t; |
123 | rethrow(t); |
124 | } |
125 | finally { |
126 | ExitStatus status = ExitStatus.FAILED; |
127 | if (startedCount == 0) { |
128 | if (steps.size() > 0) { |
129 | status = ExitStatus.NOOP |
130 | .addExitDescription("All steps already completed. No processing was done."); |
131 | } |
132 | else { |
133 | status = ExitStatus.NOOP.addExitDescription("No steps configured for this job."); |
134 | } |
135 | } |
136 | else if (currentStepExecution != null) { |
137 | status = currentStepExecution.getExitStatus(); |
138 | } |
139 | execution.setExitStatus(status); |
140 | |
141 | BatchStatus jobStatus = execution.getStatus(); |
142 | if (jobStatus == BatchStatus.COMPLETED) { |
143 | getCompositeListener().afterJob(execution); |
144 | } |
145 | else if (jobStatus == BatchStatus.FAILED) { |
146 | getCompositeListener().onError(execution, jobFailure); |
147 | } |
148 | else if (jobStatus == BatchStatus.STOPPED) { |
149 | getCompositeListener().onInterrupt(execution); |
150 | } |
151 | |
152 | execution.setEndTime(new Date()); |
153 | getJobRepository().saveOrUpdate(execution); |
154 | } |
155 | |
156 | } |
157 | |
158 | private void updateStatus(JobExecution jobExecution, BatchStatus status) { |
159 | jobExecution.setStatus(status); |
160 | getJobRepository().saveOrUpdate(jobExecution); |
161 | } |
162 | |
163 | /* |
164 | * Given a step and configuration, return true if the step should start, |
165 | * false if it should not, and throw an exception if the job should finish. |
166 | */ |
167 | private boolean shouldStart(JobInstance jobInstance, Step step) throws JobExecutionException { |
168 | |
169 | BatchStatus stepStatus; |
170 | // if the last execution is null, the step has never been executed. |
171 | StepExecution lastStepExecution = getJobRepository().getLastStepExecution(jobInstance, step); |
172 | if (lastStepExecution == null) { |
173 | stepStatus = BatchStatus.STARTING; |
174 | } |
175 | else { |
176 | stepStatus = lastStepExecution.getStatus(); |
177 | } |
178 | |
179 | if (stepStatus == BatchStatus.UNKNOWN) { |
180 | throw new JobExecutionException("Cannot restart step from UNKNOWN status. " |
181 | + "The last execution ended with a failure that could not be rolled back, " |
182 | + "so it may be dangerous to proceed. " + "Manual intervention is probably necessary."); |
183 | } |
184 | |
185 | if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) { |
186 | // step is complete, false should be returned, indicating that the |
187 | // step should not be started |
188 | return false; |
189 | } |
190 | |
191 | if (getJobRepository().getStepExecutionCount(jobInstance, step) < step.getStartLimit()) { |
192 | // step start count is less than start max, return true |
193 | return true; |
194 | } |
195 | else { |
196 | // start max has been exceeded, throw an exception. |
197 | throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName() |
198 | + "StartMax: " + step.getStartLimit()); |
199 | } |
200 | } |
201 | |
202 | /** |
203 | * @param t |
204 | */ |
205 | private static void rethrow(Throwable t) throws RuntimeException { |
206 | if (t instanceof RuntimeException) { |
207 | throw (RuntimeException) t; |
208 | } |
209 | else if (t instanceof Error) { |
210 | throw (Error) t; |
211 | } |
212 | else { |
213 | throw new UnexpectedJobExecutionException("Unexpected checked exception in job execution", t); |
214 | } |
215 | } |
216 | } |