OPEN - issue BATCH-890, BATCH-879: Stop transition in XML namespace, decision state support.

Added StepExecution to decider interface and some protection for multithreaded job executions.

spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java

@@ -16,7 +16,6 @@
 
 package org.springframework.batch.core;
 
-
 /**
  * Enumeration representing the status of a an Execution.
  * 
@@ -26,16 +25,22 @@ package org.springframework.batch.core;
 
 public enum BatchStatus {
 
-	COMPLETED, STARTED, STARTING, PAUSED, FAILED, STOPPING, STOPPED, UNKNOWN;
-	
+	/**
+	 * The order of the status values is significant. An execution is expected
+	 * to move up from lowest to highest, hopefully stopping at COMPLETED. After
+	 * COMPLETED, higher values signify more serious failure.
+	 */
+	STARTING, STARTED, COMPLETED, PAUSED, FAILED, STOPPING, STOPPED, UNKNOWN;
+
 	public static BatchStatus max(BatchStatus status1, BatchStatus status2) {
-		if (status1.compareTo(status2)<0) {
+		if (status1.compareTo(status2) < 0) {
 			return status2;
 		}
-		if (status1.compareTo(status2)>0) {
+		if (status1.compareTo(status2) > 0) {
 			return status1;
 		}
-		else return status1;
+		else
+			return status1;
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java

@@ -48,13 +48,13 @@ public class JobExecution extends Entity {
 	private volatile Date createTime = new Date(System.currentTimeMillis());
 
 	private volatile Date endTime = null;
-	
+
 	private volatile Date lastUpdated = null;
 
 	private volatile ExitStatus exitStatus = ExitStatus.UNKNOWN;
 
 	private volatile ExecutionContext executionContext = new ExecutionContext();
-	
+
 	private transient volatile List<Throwable> failureExceptions = new ArrayList<Throwable>();
 
 	/**
@@ -76,8 +76,8 @@ public class JobExecution extends Entity {
 	public JobExecution(JobInstance job) {
 		this(job, null);
 	}
-	
-	public JobExecution(Long id){
+
+	public JobExecution(Long id) {
 		super(id);
 	}
 
@@ -88,7 +88,7 @@ public class JobExecution extends Entity {
 	public void setJobInstance(JobInstance jobInstance) {
 		this.jobInstance = jobInstance;
 	}
-	
+
 	public void setEndTime(Date endTime) {
 		this.endTime = endTime;
 	}
@@ -105,10 +105,28 @@ public class JobExecution extends Entity {
 		return status;
 	}
 
+	/**
+	 * Set the value of the status field.
+	 * 
+	 * @param status the status to set
+	 */
 	public void setStatus(BatchStatus status) {
 		this.status = status;
 	}
 
+	/**
+	 * Upgrade the status field if the provided value is greater than the
+	 * existing one. Clients using this method to set the status can be sure
+	 * that they don't overwrite a failed status with an successful one.
+	 * 
+	 * @param status the new status value
+	 */
+	public void upgradeStatus(BatchStatus status) {
+		if (status.compareTo(this.status) > 0) {
+			this.status = status;
+		}
+	}
+
 	/**
 	 * Convenience getter for for the id of the enclosing job. Useful for DAO
 	 * implementations.
@@ -194,12 +212,14 @@ public class JobExecution extends Entity {
 	}
 
 	/**
-	 * Signal that this job execution wishes to be paused.
+	 * Signal that this job execution wishes to be paused. Uses
+	 * {@link #upgradeStatus(BatchStatus)} so that a failed execution stays
+	 * failed.
 	 */
 	public void pause() {
-		status = BatchStatus.PAUSED;
+		upgradeStatus(BatchStatus.PAUSED);
 	}
-	
+
 	/**
 	 * Test if the {@link JobExecution} has been paused.
 	 * 
@@ -208,7 +228,7 @@ public class JobExecution extends Entity {
 	 * @return true if this instance is paused
 	 */
 	public boolean isPaused() {
-		return status==BatchStatus.PAUSED;
+		return status == BatchStatus.PAUSED;
 	}
 
 	/**
@@ -253,8 +273,8 @@ public class JobExecution extends Entity {
 	}
 
 	/**
-	 * Get the date representing the last time this JobExecution was updated in the
-	 * JobRepository.
+	 * Get the date representing the last time this JobExecution was updated in
+	 * the JobRepository.
 	 * 
 	 * @return Date representing the last time this JobExecution was updated.
 	 */
@@ -270,39 +290,40 @@ public class JobExecution extends Entity {
 	public void setLastUpdated(Date lastUpdated) {
 		this.lastUpdated = lastUpdated;
 	}
-	
+
 	public List<Throwable> getFailureExceptions() {
 		return failureExceptions;
 	}
-	
+
 	/**
-	 * Add the provided throwable to the failure exception list. 
+	 * Add the provided throwable to the failure exception list.
 	 * 
 	 * @param t
 	 */
-	public void addFailureException(Throwable t){
+	public void addFailureException(Throwable t) {
 		this.failureExceptions.add(t);
 	}
-	
+
 	/**
 	 * Return all failure causing exceptions for this JobExecution, including
 	 * step executions.
 	 * 
-	 * @return List<Throwable> containing all exceptions causing failure for this
-	 * JobExecution.
+	 * @return List<Throwable> containing all exceptions causing failure for
+	 * this JobExecution.
 	 */
-	public List<Throwable> getAllFailureExceptions(){
-		
+	public List<Throwable> getAllFailureExceptions() {
+
 		Set<Throwable> allExceptions = new HashSet<Throwable>(failureExceptions);
-		for(StepExecution stepExecution: stepExecutions){
+		for (StepExecution stepExecution : stepExecutions) {
 			allExceptions.addAll(stepExecution.getFailureExceptions());
 		}
-		
+
 		return new ArrayList<Throwable>(allExceptions);
 	}
 
 	/**
-	 * Deserialise and ensure transient fields are re-instantiated when read back
+	 * Deserialise and ensure transient fields are re-instantiated when read
+	 * back
 	 */
 	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
 		stream.defaultReadObject();
@@ -311,10 +332,13 @@ public class JobExecution extends Entity {
 
 	/*
 	 * (non-Javadoc)
+	 * 
 	 * @see org.springframework.batch.core.domain.Entity#toString()
 	 */
 	public String toString() {
-		return super.toString() + String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s]",startTime,endTime,lastUpdated,status,exitStatus,jobInstance);
+		return super.toString()
+				+ String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s]",
+						startTime, endTime, lastUpdated, status, exitStatus, jobInstance);
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/DecisionState.java

@@ -20,7 +20,7 @@ public class DecisionState extends AbstractState<JobFlowExecutor> {
 
 	@Override
 	public String handle(JobFlowExecutor context) throws Exception {
-		return decider.decide(context.getJobExecution());
+		return decider.decide(context.getJobExecution(), context.getStepExecution());
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/EndState.java

@@ -38,7 +38,7 @@ public class EndState extends AbstractState<JobFlowExecutor> {
 		// If there are no step executions, then we are at the beginning of a
 		// restart
 		if (!jobExecution.getStepExecutions().isEmpty()) {
-			jobExecution.setStatus(status);
+			jobExecution.upgradeStatus(status);
 		}
 		return FlowExecution.COMPLETED;
 	}

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java

@@ -26,6 +26,7 @@ import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.flow.Flow;
 import org.springframework.batch.flow.FlowExecution;
 import org.springframework.batch.flow.FlowExecutionException;
+import org.springframework.batch.flow.FlowExecutionListenerSupport;
 import org.springframework.util.Assert;
 
 
@@ -65,15 +66,7 @@ public class FlowJob extends AbstractJob {
 	@Override
 	protected StepExecution doExecute(final JobExecution execution) throws JobExecutionException {
 		try {
-			FlowExecution result = flow.start(new JobFlowExecutor() {
-				public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException {
-					StepExecution stepExecution = handleStep(step, execution);
-					return stepExecution==null ? FlowExecution.COMPLETED : stepExecution.getExitStatus().getExitCode();
-				}
-				public JobExecution getJobExecution() {
-					return execution;
-				}
-			});
+			FlowExecution result = flow.start(new SimpleJobFlowExecutor(execution));
 			return getLastStepExecution(execution, result);
 		}
 		catch (FlowExecutionException e) {
@@ -128,4 +121,42 @@ public class FlowJob extends AbstractJob {
 		return first.getEndTime().after(second.getEndTime());
 	}
 
+	/**
+	 * @author Dave Syer
+	 *
+	 */
+	private class SimpleJobFlowExecutor extends FlowExecutionListenerSupport implements JobFlowExecutor {
+
+		private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<StepExecution>();
+		private final JobExecution execution;
+
+		/**
+		 * @param execution
+		 */
+		private SimpleJobFlowExecutor(JobExecution execution) {
+			this.execution = execution;
+			stepExecutionHolder.set(null);
+		}
+
+		public String executeStep(Step step) throws JobInterruptedException, JobRestartException, StartLimitExceededException {
+			StepExecution stepExecution = handleStep(step, execution);
+			stepExecutionHolder.set(stepExecution);
+			return stepExecution==null ? FlowExecution.COMPLETED : stepExecution.getExitStatus().getExitCode();
+		}
+
+		public JobExecution getJobExecution() {
+			return execution;
+		}
+		
+		public StepExecution getStepExecution() {
+			return stepExecutionHolder.get();
+		}
+		
+		@Override
+		public void close(FlowExecution result) {
+			stepExecutionHolder.set(null);
+		}
+
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobExecutionDecider.java

@@ -16,6 +16,7 @@
 package org.springframework.batch.core.job.flow;
 
 import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
 
 /**
  * @author Dave Syer
@@ -29,8 +30,9 @@ public interface JobExecutionDecider {
 	 * determine the next step in the job.
 	 * 
 	 * @param jobExecution a job execution
+	 * @param stepExecution the latest step execution (may be null)
 	 * @return the exit status code
 	 */
-	String decide(JobExecution jobExecution);
+	String decide(JobExecution jobExecution, StepExecution stepExecution);
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java

@@ -19,6 +19,7 @@ import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInterruptedException;
 import org.springframework.batch.core.StartLimitExceededException;
 import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.batch.flow.Flow;
 
@@ -45,4 +46,9 @@ public interface JobFlowExecutor {
 	 */
 	JobExecution getJobExecution();
 
+	/**
+	 * @return the latest {@link StepExecution} or null if there is none
+	 */
+	StepExecution getStepExecution();
+
 }

spring-batch-core/src/test/java/org/springframework/batch/core/JobExecutionTests.java

@@ -92,6 +92,38 @@ public class JobExecutionTests {
 		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
 	}
 
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#getStatus()}.
+	 */
+	@Test
+	public void testUpgradeStatus() {
+		assertEquals(BatchStatus.STARTING, execution.getStatus());
+		execution.upgradeStatus(BatchStatus.COMPLETED);
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+	}
+
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#pause()}.
+	 */
+	@Test
+	public void testPause() {
+		execution.pause();
+		assertEquals(BatchStatus.PAUSED, execution.getStatus());
+	}
+
+	/**
+	 * Test method for
+	 * {@link org.springframework.batch.core.JobExecution#getStatus()}.
+	 */
+	@Test
+	public void testDowngradeStatus() {
+		execution.setStatus(BatchStatus.FAILED);
+		execution.upgradeStatus(BatchStatus.COMPLETED);
+		assertEquals(BatchStatus.FAILED, execution.getStatus());
+	}
+
 	/**
 	 * Test method for
 	 * {@link org.springframework.batch.core.JobExecution#getJobId()}.

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DecisionJobParserTests.java

@@ -25,6 +25,7 @@ import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.flow.JobExecutionDecider;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
@@ -64,7 +65,7 @@ public class DecisionJobParserTests {
 	}
 	
 	public static class TestDecider implements JobExecutionDecider {
-		public String decide(JobExecution jobExecution) {
+		public String decide(JobExecution jobExecution, StepExecution stepExecution) {
 			return "FOO";
 		}
 	}

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StopJobParserTests.java

@@ -25,6 +25,7 @@ import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.flow.JobExecutionDecider;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
@@ -67,7 +68,7 @@ public class StopJobParserTests {
 	}
 
 	public static class TestDecider implements JobExecutionDecider {
-		public String decide(JobExecution jobExecution) {
+		public String decide(JobExecution jobExecution, StepExecution stepExecution) {
 			return "FOO";
 		}
 	}

spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/EndStateTests.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2006-2007 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
+ *
+ *      http://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.batch.core.job.flow;
+
+import static org.junit.Assert.*;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class EndStateTests {
+
+	private JobExecution jobExecution;
+	
+	@Before
+	public void setUp() {
+		jobExecution = new JobExecution(0L);
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleRestartSunnyDay() throws Exception {
+
+		BatchStatus status = jobExecution.getStatus();
+	
+		EndState state = new EndState(BatchStatus.UNKNOWN, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		assertEquals(status, jobExecution.getStatus());
+
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleOngoingSunnyDay() throws Exception {
+
+		jobExecution.createStepExecution("foo");
+	
+		EndState state = new EndState(BatchStatus.UNKNOWN, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		assertEquals(BatchStatus.UNKNOWN, jobExecution.getStatus());
+
+	}
+
+	/**
+	 * Test method for {@link EndState#handle(JobFlowExecutor)}.
+	 * @throws Exception 
+	 */
+	@Test
+	public void testHandleOngoingAttemptedDowngrade() throws Exception {
+
+		jobExecution.setStatus(BatchStatus.FAILED);
+		jobExecution.createStepExecution("foo");
+	
+		EndState state = new EndState(BatchStatus.COMPLETED, "end");
+		state.handle(new JobFlowExecutorSupport() {
+			@Override
+			public JobExecution getJobExecution() {
+				return jobExecution;
+			}
+		});
+		
+		// Can't downgrade a status - if it failed then it failed
+		assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
+
+	}
+
+}

spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java

@@ -16,6 +16,7 @@
 package org.springframework.batch.core.job.flow;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.fail;
 
 import java.util.ArrayList;
@@ -214,12 +215,15 @@ public class FlowJobTests {
 
 	@Test
 	public void testDecisionFlow() throws Throwable {
+
 		SimpleFlow<JobFlowExecutor> flow = new SimpleFlow<JobFlowExecutor>("job");
 		JobExecutionDecider decider = new JobExecutionDecider() {
-			public String decide(JobExecution jobExecution) {
+			public String decide(JobExecution jobExecution, StepExecution stepExecution) {
+				assertNotNull(stepExecution);
 				return "SWITCH";
 			}
 		};
+
 		Collection<StateTransition<JobFlowExecutor>> transitions = new ArrayList<StateTransition<JobFlowExecutor>>();
 		transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "*", "decision"));
 		transitions.add(StateTransition.createStateTransition(new DecisionState(decider, "decision"), "*", "step2"));
@@ -228,14 +232,17 @@ public class FlowJobTests {
 		transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step2")), "*"));
 		transitions.add(StateTransition.createEndStateTransition(new StepState(new StubStep("step3")), "*"));
 		flow.setStateTransitions(transitions);
+
 		job.setFlow(flow);
 		StepExecution stepExecution = job.doExecute(jobExecution);
 		if (!jobExecution.getAllFailureExceptions().isEmpty()) {
 			throw jobExecution.getAllFailureExceptions().get(0);
 		}
+
 		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
 		assertEquals(2, jobExecution.getStepExecutions().size());
 		assertEquals("step3", stepExecution.getStepName());
+
 	}
 
 	@Test

spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/JobFlowExecutorSupport.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2006-2007 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
+ *
+ *      http://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.batch.core.job.flow;
+
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.StartLimitExceededException;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.repository.JobRestartException;
+import org.springframework.batch.flow.FlowExecution;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class JobFlowExecutorSupport implements JobFlowExecutor {
+
+	public String executeStep(Step step) throws JobInterruptedException, JobRestartException,
+			StartLimitExceededException {
+		return FlowExecution.COMPLETED;
+	}
+
+	public JobExecution getJobExecution() {
+		return null;
+	}
+
+	public StepExecution getStepExecution() {
+		return null;
+	}
+
+}