CLOSED - issue BATCH-136: restoreFromRestartData(Tasklet, RestartData) crashes if Properties are null

http://opensource.atlassian.com/projects/spring/browse/BATCH-136

execution/src/main/java/org/springframework/batch/execution/bootstrap/support/BatchCommandLineLauncher.java

@@ -32,6 +32,50 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.util.Assert;
 
 /**
+ *<p>Basic Launcher for starting jobs from the command line.  In general,
+ * it is assumed that this launcher will primarily be used to start
+ * a job via a script from an Enterprise Scheduler.  Therefore, exit codes
+ * are mapped to integers so that schedulers can use the returned values to
+ * determine the next course of action.  The returned values can also
+ * be useful to operations teams in determining what should happen upon failure.
+ * For example, a returned code of 5 might mean that some resource wasn't available
+ * and the job should be restarted.  However, a code of 10 might mean that something
+ * critical has happened and the issue should be escalated.</p>
+ *
+ * <p>With any launch of a batch job within Spring Batch, a minimum of two contexts
+ * must be loaded.  One is the context containing the JobConfiguration, the other
+ * contains the 'Execution Environment'.  That is, the JobExecutorFacade (which
+ * contains all the executors, plus the repository), the JobIdentifierFactory, and
+ * a normal JobLauncher.  This command line launcher loads these application contexts
+ * by first loading the execution environment context via a
+ * {@link ContextSingletonBeanFactoryLocator}, which will search for the default
+ * key from classpath*:beanRefContext.xml to return the context.  This will then
+ * be used as the parent to the JobConfiguration context.  All required dependencies
+ * of the launcher will then be satisfied by autowiring by type from the combined
+ * application context.  Default values are provided for all fields except the JobLauncher.
+ * Therefore, if autowiring fails to set it (it should be noted that dependency checking
+ * is disabled because most of the fields have default values and thus don't require
+ * dependencies to be fulfilled via autowiring) then an exception will be thrown.
+ * It should be noted that even if an exception is thrown by this class, it will be
+ * mapped to an integer and returned.</p>
+ *
+ * <p>One odd field might be noticed in the launcher, SystemExiter.  This class is
+ * used to exit from the main method, rather than calling System.exit directly.  This
+ * is because unit testing a class the calls System.exit() is impossible without
+ * kicking off the test within a new Jvm, which it is possible to do, however it is a
+ * complex solution, much more so than strategizing the exiter.</p>
+ *
+ * <p>VM Arguments vs. Program arguments:  Because all of the arguments to the main
+ * method are optional, VM arguments are used:
+ *
+ * <ul>
+ * 	<li>-Djob.configuration.path: the classpath location of the JobConfiguration
+ * to use
+ * 	<li>-Djob.name: job name to be passed to the {@link JobLauncher}
+ *  <li>-Dbatch.execution.environment.key: the key in beanRefContext.xml used to load
+ *  the execution envrionement.
+ * </ul>
+ *
  * @author Dave Syer
  * @author Lucas Ward
  * @since 2.1
@@ -113,6 +157,8 @@ public class BatchCommandLineLauncher {
 	 *            the path to a Spring context configuration for this job
 	 * @param jobName
 	 *            the name of the job execution to use
+	 * @parm parentKey the key to be loaded by ContextSingletonBeanFactoryLocator and
+	 * used as the parent context.
 	 * @throws NoSuchJobConfigurationException
 	 * @throws IllegalStateException
 	 *             if JobLauncher is not autowired by the ApplicationContext
@@ -171,18 +217,17 @@ public class BatchCommandLineLauncher {
 	/**
 	 * Launch a batch job using a {@link BatchCommandLineLauncher}. Creates a
 	 * new Spring context for the job execution, and uses a common parent for
-	 * all such contexts.
+	 * all such contexts.  No exception should be thrown from this method, rather
+	 * exceptions should be logged and an integer returned.
 	 *
 	 * @param args
-	 *            <ol>
-	 *            <li> classpath location of resource to load job configuration
-	 *            context (default "job-configuration.xml");</li>
-	 *            <li>runtime name of Job (default "job-execution-id").</li>
-	 *            <li> parent context key for use in pulling the correct
-	 *            beanFactory from the beanRefContext.xml (@see
-	 *            ContextSingletonBeanFactoryLocator)</li>
-	 *            </ol>
-	 * @throws NoSuchJobConfigurationException
+	 * <ul>
+	 * 	<li>-Djob.configuration.path: the classpath location of the JobConfiguration
+	 * 		to use
+	 * 	<li>-Djob.name: job name to be passed to the {@link JobLauncher}
+	 *  <li>-Dbatch.execution.environment.key: the key in beanRefContext.xml used to load
+	 *  the execution envrionement.
+	 * </ul>
 	 */
 	public static void main(String[] args) {
 

execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java

@@ -19,6 +19,7 @@ package org.springframework.batch.execution.repository;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Properties;
 
 import org.springframework.batch.core.configuration.JobConfiguration;
 import org.springframework.batch.core.configuration.StepConfiguration;
@@ -32,21 +33,22 @@ import org.springframework.batch.core.repository.NoSuchBatchDomainObjectExceptio
 import org.springframework.batch.core.runtime.JobIdentifier;
 import org.springframework.batch.execution.repository.dao.JobDao;
 import org.springframework.batch.execution.repository.dao.StepDao;
+import org.springframework.batch.restart.GenericRestartData;
 import org.springframework.util.Assert;
 
 /**
- * 
+ *
  * <p>
  * Simple Job Repository that stores Jobs, JobExecutions, Steps, and
  * StepExecutions using the provided JobDao and StepDao.
  * <p>
- * 
+ *
  * @author Lucas Ward
  * @author Dave Syer
  * @see JobRepository
  * @see StepDao
  * @see JobDao
- * 
+ *
  */
 public class SimpleJobRepository implements JobRepository {
 
@@ -62,15 +64,15 @@ public class SimpleJobRepository implements JobRepository {
 
 	/**
 	 * <p>
-	 * Find or Create a Job(@link Job) based on the passed in RuntimeInformation
-	 * and Configuration. JobRuntimeInformation contains the following fields
-	 * which logically identify a job: JobName, JobStream, JobRun, and Schedule
-	 * Date. However, unique identification of a job can only come from the
-	 * database, and therefore must come from JobDao by either creating a new
-	 * job or finding an existing one, which will ensure that the id field of
-	 * the job is populated with the correct value.
+	 * Find or Create a JobInstance(@link JobInstance) based on the passed in
+	 * RuntimeInformation and Configuration. JobRuntimeInformation contains the
+	 * following fields which logically identify a job: JobName, JobStream,
+	 * JobRun, and Schedule Date. However, unique identification of a job can
+	 * only come from the database, and therefore must come from JobDao by
+	 * either creating a new job or finding an existing one, which will ensure
+	 * that the id field of the job is populated with the correct value.
 	 * </p>
-	 * 
+	 *
 	 * <p>
 	 * There are two ways in which the method determines if a job should be
 	 * created or an existing one should be returned. The first is
@@ -81,9 +83,8 @@ public class SimpleJobRepository implements JobRepository {
 	 * (there must be at least 1) and it will be returned. If no job is found, a
 	 * new one will be created based on the configuration.
 	 * </p>
-	 * 
-	 * @see JobRepository#findOrCreateJob(JobConfiguration,
-	 * JobIdentifier)
+	 *
+	 * @see JobRepository#findOrCreateJob(JobConfiguration, JobIdentifier)
 	 */
 	public JobInstance findOrCreateJob(JobConfiguration jobConfiguration, JobIdentifier runtimeInformation) {
 
@@ -126,7 +127,7 @@ public class SimpleJobRepository implements JobRepository {
 	 * identifer, because it must be updatable separately. If an id isn't found,
 	 * a new JobExecution is created, if one is found, the current row is
 	 * updated.
-	 * 
+	 *
 	 * @param JobExecution to be stored.
 	 * @throws IllegalArgumentException if jobExecution is null.
 	 */
@@ -149,7 +150,7 @@ public class SimpleJobRepository implements JobRepository {
 	 * Update an existing job. A job must have been obtained from the
 	 * findOrCreateJob method, otherwise it is likely that the id is incorrect
 	 * or non-existant.
-	 * 
+	 *
 	 * @param job to be updated.
 	 * @throws IllegalArgumentException if Job or it's Id is null.
 	 */
@@ -167,7 +168,7 @@ public class SimpleJobRepository implements JobRepository {
 	 * saved and an id will be set, otherwise it will be updated. It should be
 	 * noted that assigning an ID randomly will likely cause an exception
 	 * depending on the StepDao implementation.
-	 * 
+	 *
 	 * @param StepExecution to be saved.
 	 * @throws IllegalArgumentException if stepExecution is null.
 	 */
@@ -188,7 +189,7 @@ public class SimpleJobRepository implements JobRepository {
 
 	/**
 	 * Update the given step.
-	 * 
+	 *
 	 * @param StepInstance to be updated.
 	 * @throws IllegalArgumentException if step or it's id is null.
 	 */
@@ -224,6 +225,10 @@ public class SimpleJobRepository implements JobRepository {
 		while (i.hasNext()) {
 			StepConfiguration stepConfiguration = (StepConfiguration) i.next();
 			StepInstance step = stepDao.createStep(job, stepConfiguration.getName());
+			//Ensure valid restart data is being returned.
+			if(step.getRestartData() == null || step.getRestartData().getProperties() == null){
+				step.setRestartData(new GenericRestartData(new Properties()));
+			}
 			steps.add(step);
 		}
 
@@ -243,7 +248,10 @@ public class SimpleJobRepository implements JobRepository {
 			if (step != null) {
 
 				step.setStepExecutionCount(stepDao.getStepExecutionCount(step.getId()));
-
+				//Ensure valid restart data is being returned.
+				if(step.getRestartData() == null || step.getRestartData().getProperties() == null){
+					step.setRestartData(new GenericRestartData(new Properties()));
+				}
 				steps.add(step);
 			}
 		}

execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java

@@ -61,15 +61,15 @@ import org.springframework.util.Assert;
  * operations should not do any concurrent execution. N.B. usually that means
  * that the chunk operations should be a {@link RepeatTemplate} (which is the
  * default).<br/>
- * 
+ *
  * Clients can use interceptors in the step operations to intercept or listen to
  * the iteration on a step-wide basis, for instance to get a callback when the
  * step is complete. Those that want callbacks at the level of an individual
  * tasks, can specify interceptors for the chunk operations.
- * 
+ *
  * @author Dave Syer
  * @author Lucas Ward
- * 
+ *
  */
 public class SimpleStepExecutor implements StepExecutor {
 
@@ -95,7 +95,7 @@ public class SimpleStepExecutor implements StepExecutor {
 	private RepeatOperations stepOperations = new RepeatTemplate();
 
 	private JobRepository jobRepository;
-	
+
 	private ExitCodeExceptionClassifier exceptionClassifier = new SimpleExitCodeExceptionClassifier();
 
 	// default to checking current thread for interruption.
@@ -155,6 +155,7 @@ public class SimpleStepExecutor implements StepExecutor {
 			throws BatchCriticalException, StepInterruptedException {
 
 		final StepInstance step = stepExecutionContext.getStep();
+		boolean isRestart = step.getStepExecutionCount() > 0 ? true : false;
 		Assert.notNull(step);
 
 		final StepExecution stepExecution = stepExecutionContext.getStepExecution();
@@ -171,7 +172,7 @@ public class SimpleStepExecutor implements StepExecutor {
 
 			final boolean saveRestartData = ((AbstractStepConfiguration) configuration).isSaveRestartData();
 
-			if (saveRestartData) {
+			if (saveRestartData && isRestart) {
 				restoreFromRestartData(module, step.getRestartData());
 			}
 
@@ -253,7 +254,7 @@ public class SimpleStepExecutor implements StepExecutor {
 			return status;
 		}
 		catch (RuntimeException e) {
-			
+
 			//classify exception so an exit code can be stored.
 			status = exceptionClassifier.classifyForExitCode(e);
 			stepExecution.setException(e);
@@ -265,7 +266,7 @@ public class SimpleStepExecutor implements StepExecutor {
 				updateStatus(stepExecutionContext, BatchStatus.FAILED);
 				throw e;
 			}
-			
+
 		}
 		finally {
 			stepExecution.setExitCode(status.getExitCode());
@@ -311,7 +312,7 @@ public class SimpleStepExecutor implements StepExecutor {
 	 * transaction. The transaction is programmatically started and stopped
 	 * outside this method, so subclasses that override do not need to create a
 	 * transaction.
-	 * 
+	 *
 	 * @param configuration the current step configuration
 	 * @param stepExecutionContext the current step, containing the
 	 * {@link Tasklet} with the business logic.
@@ -384,11 +385,11 @@ public class SimpleStepExecutor implements StepExecutor {
 	public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) {
 		this.interruptionPolicy = interruptionPolicy;
 	}
-	
+
 	/**
 	 * Setter for the {@link ExitCodeExceptionClassifier} that will be used
 	 * to classify any exception that causes a job to fail.
-	 * 
+	 *
 	 * @param exceptionClassifier
 	 */
 	public void setExceptionClassifier(

execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java

@@ -33,12 +33,13 @@ import org.springframework.batch.core.runtime.SimpleJobIdentifier;
 import org.springframework.batch.core.tasklet.Tasklet;
 import org.springframework.batch.execution.repository.dao.JobDao;
 import org.springframework.batch.execution.repository.dao.StepDao;
+import org.springframework.batch.restart.GenericRestartData;
 
 /*
  * Test SimpleJobRepository.  The majority of test cases are tested using EasyMock,
- * however, there were some issues with using it for the stepDao when testing finding 
- * or creating steps, so an actual mock class had to be written. 
- * 
+ * however, there were some issues with using it for the stepDao when testing finding
+ * or creating steps, so an actual mock class had to be written.
+ *
  * @author Lucas Ward
  *
  */
@@ -69,14 +70,14 @@ public class SimpleJobRepositoryTests extends TestCase {
 	StepInstance databaseStep1;
 
 	StepInstance databaseStep2;
-	
+
 	List steps;
 
 	public void setUp() throws Exception {
 
 		jobDao = (JobDao) jobDaoControl.getMock();
 		stepDao = (StepDao) stepDaoControl.getMock();
-		
+
 		jobRepository = new SimpleJobRepository(jobDao, stepDao);
 
 		jobRuntimeInformation = new SimpleJobIdentifier("RepositoryTest");
@@ -99,14 +100,17 @@ public class SimpleJobRepositoryTests extends TestCase {
 
 		databaseStep1 = new StepInstance(new Long(1));
 		databaseStep2 = new StepInstance(new Long(2));
-		
+
 		steps = new ArrayList();
 		steps.add(databaseStep1);
 		steps.add(databaseStep2);
 	}
-	
+
+	/*
+	 * Test a restartable job, that has not been run before.
+	 */
 	public void testCreateRestartableJob(){
-	
+
 		List jobs = new ArrayList();
 
 		jobDao.findJobs(jobRuntimeInformation);
@@ -128,7 +132,7 @@ public class SimpleJobRepositoryTests extends TestCase {
 		step = (StepInstance) it.next();
 		assertTrue(step.equals(databaseStep2));
 	}
-	
+
 	public void testRestartedJob(){
 		List jobs = new ArrayList();
 		jobDao.findJobs(jobRuntimeInformation);
@@ -157,9 +161,9 @@ public class SimpleJobRepositoryTests extends TestCase {
 		assertTrue(step.equals(databaseStep2));
 		assertTrue(step.getStepExecutionCount() == 1);
 	}
-	
+
 	public void testCreateNonRestartableJob(){
-		
+
 		List jobs = new ArrayList();
 
 		jobDao.findJobs(jobRuntimeInformation);
@@ -247,7 +251,7 @@ public class SimpleJobRepositoryTests extends TestCase {
 		stepDaoControl.replay();
 		jobRepository.update(step);
 	}
-	
+
 	public void testUpdateStepExecution(){
 		StepExecution stepExecution = new StepExecution(new Long(10), null);
 		stepExecution.setId(new Long(11));
@@ -256,10 +260,10 @@ public class SimpleJobRepositoryTests extends TestCase {
 		jobRepository.saveOrUpdate(stepExecution);
 		stepDaoControl.verify();
 	}
-	
+
 	public void testSaveStepExecution(){
 		StepExecution stepExecution = new StepExecution(new Long(10), null);
-		//TODO: Not sure why, but calling save on the EasyMock stepDao causes a NullPointerException 
+		//TODO: Not sure why, but calling save on the EasyMock stepDao causes a NullPointerException
 //		stepDao.save(stepExecution);
 //		stepDaoControl.replay();
 		jobRepository.saveOrUpdate(stepExecution);
@@ -280,6 +284,68 @@ public class SimpleJobRepositoryTests extends TestCase {
 		}
 	}
 
+	/*
+	 * Test to ensure that if a StepDao returns invalid
+	 * restart data, it is corrected.
+	 */
+	public void testCreateStepsFixesInvalidRestartData(){
+
+		List jobs = new ArrayList();
+
+		jobDao.findJobs(jobRuntimeInformation);
+		jobDaoControl.setReturnValue(jobs);
+		jobDao.createJob(jobRuntimeInformation);
+		jobDaoControl.setReturnValue(databaseJob);
+		stepDao.createStep(databaseJob, "TestStep1");
+		databaseStep1.setRestartData(null);
+		stepDaoControl.setReturnValue(databaseStep1);
+		stepDao.createStep(databaseJob, "TestStep2");
+		databaseStep2.setRestartData(new GenericRestartData(null));
+		stepDaoControl.setReturnValue(databaseStep2);
+		stepDaoControl.replay();
+		jobDaoControl.replay();
+		JobInstance job = jobRepository.findOrCreateJob(jobConfiguration, jobRuntimeInformation);
+		List jobSteps = job.getSteps();
+		Iterator it = jobSteps.iterator();
+		StepInstance step = (StepInstance) it.next();
+		assertTrue(step.equals(databaseStep1));
+		assertTrue(step.getRestartData().getProperties().isEmpty());
+		step = (StepInstance) it.next();
+		assertTrue(step.equals(databaseStep2));
+		assertTrue(step.getRestartData().getProperties().isEmpty());
+	}
+
+	public void testFindStepsFixesInvalidRestartData(){
+		List jobs = new ArrayList();
+		jobDao.findJobs(jobRuntimeInformation);
+		jobs.add(databaseJob);
+		jobDaoControl.setReturnValue(jobs);
+		stepDao.findStep(databaseJob, "TestStep1");
+		databaseStep1.setRestartData(null);
+		stepDaoControl.setReturnValue(databaseStep1);
+		stepDao.getStepExecutionCount(databaseStep1.getId());
+		stepDaoControl.setReturnValue(1);
+		stepDao.findStep(databaseJob, "TestStep2");
+		databaseStep2.setRestartData(new GenericRestartData(null));
+		stepDaoControl.setReturnValue(databaseStep2);
+		stepDao.getStepExecutionCount(databaseStep2.getId());
+		stepDaoControl.setReturnValue(1);
+		stepDaoControl.replay();
+		jobDao.getJobExecutionCount(databaseJob.getId());
+		jobDaoControl.setReturnValue(1);
+		jobDaoControl.replay();
+		JobInstance job = jobRepository.findOrCreateJob(jobConfiguration, jobRuntimeInformation);
+		assertTrue(job.equals(databaseJob));
+		List jobSteps = job.getSteps();
+		Iterator it = jobSteps.iterator();
+		StepInstance step = (StepInstance) it.next();
+		assertTrue(step.equals(databaseStep1));
+		assertTrue(step.getRestartData().getProperties().isEmpty());
+		step = (StepInstance) it.next();
+		assertTrue(step.getRestartData().getProperties().isEmpty());
+		assertTrue(step.equals(databaseStep2));
+	}
+
 	/**
 	 * @author Dave Syer
 	 *

execution/src/test/java/org/springframework/batch/execution/step/simple/DefaultStepExecutorTests.java

@@ -42,6 +42,8 @@ import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
+import org.springframework.batch.restart.RestartData;
+import org.springframework.batch.restart.Restartable;
 
 public class DefaultStepExecutorTests extends TestCase {
 
@@ -60,7 +62,7 @@ public class DefaultStepExecutorTests extends TestCase {
 	private ItemProvider getProvider(String[] args) {
 		return new ListItemProvider(Arrays.asList(args));
 	}
-	
+
 	/**
 	 * @param strings
 	 * @return
@@ -131,7 +133,7 @@ public class DefaultStepExecutorTests extends TestCase {
 		step.setStepExecution(new StepExecution(new Long(1),new Long(1)));
 		final JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
 		final StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
-		
+
 		stepConfiguration.setTasklet(new Tasklet() {
 			public ExitStatus execute() throws Exception {
 				assertEquals(step, stepExecutionContext.getStep());
@@ -159,66 +161,66 @@ public class DefaultStepExecutorTests extends TestCase {
 
 		JobInstance job = new JobInstance(new Long(1));
 		job.setIdentifier(new SimpleJobIdentifier("foo_bar"));
-		
+
 		stepExecutor.process(stepConfiguration, stepExecutionContext);
 		assertEquals(1, processed.size());
 
 		// assertEquals(1, repository.findJobs(job.?).size());
 	}
-	
+
 	public void testIncrementRollbackCount(){
-		
+
 		Tasklet tasklet = new Tasklet(){
 
 			public ExitStatus execute() throws Exception {
 				int counter = 0;
 				counter++;
-				
+
 				if(counter == 1){
 					throw new Exception();
 				}
-				
+
 				return ExitStatus.CONTINUABLE;
 			}
-			
+
 		};
-		
+
 		StepInstance step = new StepInstance(new Long(1));
 		stepConfiguration.setTasklet(tasklet);
 		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
 		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
-		
+
 		try{
 			stepExecutor.process(stepConfiguration, stepExecutionContext);
 		}
 		catch(Exception ex){
 			assertEquals(step.getStepExecution().getRollbackCount(), new Integer(1));
 		}
-		
+
 	}
-	
+
 	public void testExitCodeDefaultClassification(){
-		
+
 		Tasklet tasklet = new Tasklet(){
 
 			public ExitStatus execute() throws Exception {
 				int counter = 0;
 				counter++;
-				
+
 				if(counter == 1){
 					throw new RuntimeException();
 				}
-				
+
 				return ExitStatus.CONTINUABLE;
 			}
-			
+
 		};
-		
+
 		StepInstance step = new StepInstance(new Long(1));
 		stepConfiguration.setTasklet(tasklet);
 		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
 		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
-		
+
 		try{
 			stepExecutor.process(stepConfiguration, stepExecutionContext);
 		}
@@ -228,44 +230,163 @@ public class DefaultStepExecutorTests extends TestCase {
 		}
 	}
 
+
+	/*
+	 * make sure a job that has never been executed before, but does have
+	 * saveRestartData = true, doesn't have restoreFrom called on it.
+	 */
+	public void testNonRestartedJob(){
+		StepInstance step = new StepInstance(new Long(1));
+		MockRestartableTasklet tasklet = new MockRestartableTasklet();
+		stepConfiguration.setTasklet(tasklet);
+		stepConfiguration.setSaveRestartData(true);
+		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
+		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
+
+		try{
+			stepExecutor.process(stepConfiguration, stepExecutionContext);
+		}catch(Throwable t){
+			fail();
+		}
+
+		assertFalse(tasklet.isRestoreFromCalled());
+		assertTrue(tasklet.isGetRestartDataCalled());
+	}
+
+	/*
+	 * make sure a job that has been executed before, and is therefore being restarted,
+	 * is restored.
+	 */
+	public void testRestartedJob(){
+		StepInstance step = new StepInstance(new Long(1));
+		step.setStepExecutionCount(1);
+		MockRestartableTasklet tasklet = new MockRestartableTasklet();
+		stepConfiguration.setTasklet(tasklet);
+		stepConfiguration.setSaveRestartData(true);
+		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
+		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
+
+		try{
+			stepExecutor.process(stepConfiguration, stepExecutionContext);
+		}catch(Throwable t){
+			fail();
+		}
+
+		assertTrue(tasklet.isRestoreFromCalled());
+		assertTrue(tasklet.isGetRestartDataCalled());
+	}
+
+	/*
+	 * Test that a job that is being restarted, but has saveRestartData
+	 * set to false, doesn't have restore or getRestartData called on it.
+	 */
+	public void testNoSaveRestartDataRestartableJob(){
+		StepInstance step = new StepInstance(new Long(1));
+		step.setStepExecutionCount(1);
+		MockRestartableTasklet tasklet = new MockRestartableTasklet();
+		stepConfiguration.setTasklet(tasklet);
+		stepConfiguration.setSaveRestartData(false);
+		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
+		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
+
+		try{
+			stepExecutor.process(stepConfiguration, stepExecutionContext);
+		}catch(Throwable t){
+			fail();
+		}
+
+		assertFalse(tasklet.isRestoreFromCalled());
+		assertFalse(tasklet.isGetRestartDataCalled());
+	}
+
+	/*
+	 * Even though the job is restarted, and saveRestartData is true,
+	 * nothing will be restored because the Tasklet does not implement
+	 * Restartable.
+	 */
+	public void testRestartJobOnNonRestartableTasklet(){
+		StepInstance step = new StepInstance(new Long(1));
+		step.setStepExecutionCount(1);
+		stepConfiguration.setTasklet(new Tasklet(){
+			public ExitStatus execute() throws Exception {
+			return ExitStatus.FINISHED;
+		}});
+		stepConfiguration.setSaveRestartData(true);
+		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
+		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
+
+		try{
+			stepExecutor.process(stepConfiguration, stepExecutionContext);
+		}catch(Throwable t){
+			fail();
+		}
+	}
+
+	private class MockRestartableTasklet implements Tasklet, Restartable{
+
+		private boolean getRestartDataCalled = false;
+		private boolean restoreFromCalled = false;
+
+		public ExitStatus execute() throws Exception {
+			return ExitStatus.FINISHED;
+		}
+
+		public RestartData getRestartData() {
+			getRestartDataCalled = true;
+			return null;
+		}
+
+		public void restoreFrom(RestartData data) {
+			restoreFromCalled = true;
+		}
+
+		public boolean isGetRestartDataCalled() {
+			return getRestartDataCalled;
+		}
+
+		public boolean isRestoreFromCalled() {
+			return restoreFromCalled;
+		}
+	}
+
 	/*
 	 * StepExecutor will never pass StepInterruptedException to the exceptionClassifier.
-	 * This may or may not stay the same, so the test will remain commented out 
+	 * This may or may not stay the same, so the test will remain commented out
 	 * for reference purposes.
 	 */
 /*	public void testExitCodeInterruptedClassification(){
-		
+
 		StepInterruptionPolicy interruptionPolicy = new StepInterruptionPolicy(){
 
 			public void checkInterrupted(RepeatContext context)
 					throws StepInterruptedException {
 				throw new StepInterruptedException("");
 			}
-			
+
 		};
-		
+
 		stepExecutor.setInterruptionPolicy(interruptionPolicy);
-		
+
 		Tasklet tasklet = new Tasklet(){
 
 			public ExitStatus execute() throws Exception {
 				int counter = 0;
 				counter++;
-				
+
 				if(counter == 1){
 					throw new StepInterruptedException("");
 				}
-				
+
 				return ExitStatus.CONTINUABLE;
 			}
-			
+
 		};
-		
+
 		StepInstance step = new StepInstance(new Long(1));
 		stepConfiguration.setTasklet(tasklet);
 		JobExecutionContext jobExecutionContext = new JobExecutionContext(new SimpleJobIdentifier("FOO"), new JobInstance(new Long(3)));
 		StepExecutionContext stepExecutionContext = new StepExecutionContext(jobExecutionContext, step);
-		
+
 		try{
 			stepExecutor.process(stepConfiguration, stepExecutionContext);
 		}