OPEN - issue BATCH-324: Step as factory for StepExecutor

http://jira.springframework.org/browse/BATCH-324 Same treatment for JobExecutor - roll it up into the Job interface
2008-01-29 18:07:57 +00:00
parent 3d7f40ca90
commit 911e368a71
64 changed files with 486 additions and 581 deletions
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/bootstrap/support/CommandLineJobRunner.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/bootstrap/support/CommandLineJobRunner.java
@@ -20,6 +20,7 @@ import java.util.Properties;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.repository.JobLocator;
@@ -80,7 +81,7 @@ import org.springframework.util.StringUtils;
 * 
 * <p>
 * <ul>
- * <li>jobPath: the xml application context containing a {@link Job}
+ * <li>jobPath: the xml application context containing a {@link JobSupport}
 * <li>jobName: the bean id of the job.
 * <li>jobLauncherPath: the xml application context containing a
 * {@link JobLauncher}
@@ -106,7 +107,7 @@ import org.springframework.util.StringUtils;
 * none is contained in the bean factory (it searches by type) then a 
 * {@link BeanDefinitionStoreException} will be thrown.  The same exception will also
 * be thrown if there is more than one present.  Assuming the JobLauncher has been
- * set correctly, the jobName argument will be used to obtain an actual {@link Job}.
+ * set correctly, the jobName argument will be used to obtain an actual {@link JobSupport}.
 * If a {@link JobLocator} has been set, then it will be used, if not the beanFactory
 * will be asked, using the jobName as the bean id.</p>
 * 
@@ -230,7 +231,7 @@ public class CommandLineJobRunner {
 	 *            <p>
 	 *            <ul>
 	 *            <li>jobPath: the xml application context containing a
-	 *            {@link Job}
+	 *            {@link JobSupport}
 	 *            <li>jobName: the bean id of the job.
 	 *            <li>jobLauncherPath: the xml application context containing a
 	 *            {@link JobLauncher}
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/configuration/JobRegistryBeanPostProcessor.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/configuration/JobRegistryBeanPostProcessor.java
@@ -20,6 +20,7 @@ import java.util.HashSet;
 import java.util.Iterator;

 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.repository.DuplicateJobException;
 import org.springframework.batch.core.repository.JobLocator;
 import org.springframework.batch.core.repository.JobRegistry;
@@ -31,7 +32,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.util.Assert;

 /**
- * A {@link BeanPostProcessor} that registers {@link Job} beans
+ * A {@link BeanPostProcessor} that registers {@link JobSupport} beans
 * with a {@link JobRegistry}. Include a bean of this type along
 * with your job configuration, and use the same
 * {@link JobRegistry} as a {@link JobLocator} when
@@ -66,7 +67,7 @@ public class JobRegistryBeanPostProcessor implements BeanPostProcessor, Initiali
 	}

 	/**
-	 * De-register all the {@link Job} instances that were
+	 * De-register all the {@link JobSupport} instances that were
 	 * regsistered by this post processor.
 	 * @see org.springframework.beans.factory.DisposableBean#destroy()
 	 */
@@ -79,7 +80,7 @@ public class JobRegistryBeanPostProcessor implements BeanPostProcessor, Initiali
 	}

 	/**
-	 * If the bean is an instance of {@link Job} then register it.
+	 * If the bean is an instance of {@link JobSupport} then register it.
 	 * @throws FatalBeanException if there is a
 	 * {@link DuplicateJobException}.
 	 * 
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/job/DefaultJobExecutor.java
@@ -14,7 +14,7 @@
 * limitations under the License.
 */

-package org.springframework.batch.execution.job;
+package org.springframework.batch.execution.job.simple;

 import java.util.Date;
 import java.util.Iterator;
@@ -22,13 +22,11 @@ import java.util.List;

 import org.springframework.batch.common.ExceptionClassifier;
 import org.springframework.batch.core.domain.BatchStatus;
-import org.springframework.batch.core.domain.Job;
 import org.springframework.batch.core.domain.JobExecution;
-import org.springframework.batch.core.domain.JobExecutor;
 import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepExecution;
-import org.springframework.batch.core.domain.StepExecutor;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
@@ -38,13 +36,14 @@ import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;

 /**
- * Default implementation of (@link JobExecutor} interface. Sequentially
- * executes a job by iterating it's life of steps.
+ * Simple implementation of (@link Job} interface providing the ability to run a
+ * {@link JobExecution}. Sequentially executes a job by iterating it's life of
+ * steps.
 * 
 * @author Lucas Ward
 * @author Dave Syer
 */
-public class DefaultJobExecutor implements JobExecutor {
+public class SimpleJob extends JobSupport {

 	private JobRepository jobRepository;

@@ -52,13 +51,11 @@ public class DefaultJobExecutor implements JobExecutor {

 	/**
 	 * Run the specified job by looping through the steps and delegating to the
-	 * {@link StepExecutor}.
+	 * {@link Step}.
 	 * 
-	 * @see org.springframework.batch.core.domain.JobExecutor#run(org.springframework.batch.core.domain.Job,
-	 *      org.springframework.batch.core.domain.JobExecution)
+	 * @see org.springframework.batch.core.domain.Job#run(org.springframework.batch.core.domain.JobExecution)
 	 */
-	public ExitStatus run(Job job, JobExecution execution)
-			throws BatchCriticalException {
+	public ExitStatus run(JobExecution execution) throws BatchCriticalException {

 		JobInstance jobInstance = execution.getJobInstance();
 		updateStatus(execution, BatchStatus.STARTING);
@@ -71,43 +68,44 @@ public class DefaultJobExecutor implements JobExecutor {

 			int startedCount = 0;

-			List steps = job
-					.getSteps();
-			for (Iterator i = stepInstances.iterator(), j = steps.iterator(); i.hasNext()
-					&& j.hasNext();) {
+			List steps = getSteps();
+			for (Iterator i = stepInstances.iterator(), j = steps.iterator(); i.hasNext() && j.hasNext();) {

 				StepInstance stepInstance = (StepInstance) i.next();
-				Step step = (Step) j
-						.next();
-				
+				Step step = (Step) j.next();
+
 				if (shouldStart(stepInstance, step)) {
 					startedCount++;
 					updateStatus(execution, BatchStatus.STARTED);
-					StepExecutor stepExecutor = step.createStepExecutor();
 					StepExecution stepExecution = execution.createStepExecution(stepInstance);
-					status = stepExecutor.process(stepExecution);
+					status = step.process(stepExecution);
 				}
 			}
-			
-			if (startedCount==0) {
-				if (steps.size()>0) {
-					status = ExitStatus.NOOP.addExitDescription("All steps already completed.  No processing was done.");
-				} else {
+
+			if (startedCount == 0) {
+				if (steps.size() > 0) {
+					status = ExitStatus.NOOP
+							.addExitDescription("All steps already completed.  No processing was done.");
+				}
+				else {
 					status = ExitStatus.NOOP.addExitDescription("No steps configured for this job.");
 				}
 			}

 			updateStatus(execution, BatchStatus.COMPLETED);

-		} catch (StepInterruptedException e) {
+		}
+		catch (StepInterruptedException e) {
 			updateStatus(execution, BatchStatus.STOPPED);
 			status = exceptionClassifier.classifyForExitCode(e);
 			rethrow(e);
-		} catch (Throwable t) {
+		}
+		catch (Throwable t) {
 			updateStatus(execution, BatchStatus.FAILED);
 			status = exceptionClassifier.classifyForExitCode(t);
 			rethrow(t);
-		} finally {
+		}
+		finally {
 			execution.setEndTime(new Date(System.currentTimeMillis()));
 			execution.setExitStatus(status);
 			jobRepository.saveOrUpdate(execution);
@@ -128,11 +126,9 @@ public class DefaultJobExecutor implements JobExecutor {
 	 * Given a step and configuration, return true if the step should start,
 	 * false if it should not, and throw an exception if the job should finish.
 	 */
-	private boolean shouldStart(StepInstance stepInstance,
-			Step step) {
+	private boolean shouldStart(StepInstance stepInstance, Step step) {

-		if (stepInstance.getStatus() == BatchStatus.COMPLETED
-				&& step.isAllowStartIfComplete() == false) {
+		if (stepInstance.getStatus() == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) {
 			// step is complete, false should be returned, indicated that the
 			// step should
 			// not be started
@@ -142,11 +138,11 @@ public class DefaultJobExecutor implements JobExecutor {
 		if (stepInstance.getStepExecutionCount() < step.getStartLimit()) {
 			// step start count is less than start max, return true
 			return true;
-		} else {
+		}
+		else {
 			// start max has been exceeded, throw an exception.
-			throw new BatchCriticalException(
-					"Maximum start limit exceeded for step: " + stepInstance.getName()
-							+ "StartMax: " + step.getStartLimit());
+			throw new BatchCriticalException("Maximum start limit exceeded for step: " + stepInstance.getName()
+					+ "StartMax: " + step.getStartLimit());
 		}
 	}

@@ -156,7 +152,8 @@ public class DefaultJobExecutor implements JobExecutor {
 	private static void rethrow(Throwable t) throws RuntimeException {
 		if (t instanceof RuntimeException) {
 			throw (RuntimeException) t;
-		} else {
+		}
+		else {
 			throw new BatchCriticalException(t);
 		}
 	}
@@ -178,8 +175,7 @@ public class DefaultJobExecutor implements JobExecutor {
 	 * 
 	 * @param exceptionClassifier
 	 */
-	public void setExceptionClassifier(
-			ExitCodeExceptionClassifier exceptionClassifier) {
+	public void setExceptionClassifier(ExitCodeExceptionClassifier exceptionClassifier) {
 		this.exceptionClassifier = exceptionClassifier;
 	}
 }
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/JobLauncher.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/JobLauncher.java
@@ -16,6 +16,7 @@
 package org.springframework.batch.execution.launch;

 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
@@ -35,7 +36,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
 public interface JobLauncher {

 	/**
-	 * Start a job execution for the given {@link Job} and {@link JobParameters}.
+	 * Start a job execution for the given {@link JobSupport} and {@link JobParameters}.
 	 * 
 	 * @return the exit code from the job if it returns synchronously. If the
 	 *         implementation is asynchronous, the status might well be unknown.
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/launch/SimpleJobLauncher.java
@@ -19,9 +19,9 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.domain.Job;
 import org.springframework.batch.core.domain.JobExecution;
-import org.springframework.batch.core.domain.JobExecutor;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.repeat.ExitStatus;
@@ -31,70 +31,67 @@ import org.springframework.core.task.TaskExecutor;
 import org.springframework.util.Assert;

 /**
- * Simple implementation of the {@link JobLauncher} interface.  The Spring
- * Core {@link TaskExecutor} interface is used to launch a {@link Job}.  This means
- * that the type of executor set is very important.  If a {@link SyncTaskExecutor} 
- * is used, then the job will be processed <strong>within the same thread that
- * called the launcher.</strong> Care should be taken to ensure any users of this
- * class understand fully whether or not the implementation of TaskExecutor used
- * will start tasks synchronously or asynchronously.  The default setting uses
- * a synchronous task executor.
+ * Simple implementation of the {@link JobLauncher} interface. The Spring Core
+ * {@link TaskExecutor} interface is used to launch a {@link JobSupport}. This
+ * means that the type of executor set is very important. If a
+ * {@link SyncTaskExecutor} is used, then the job will be processed
+ * <strong>within the same thread that called the launcher.</strong> Care
+ * should be taken to ensure any users of this class understand fully whether or
+ * not the implementation of TaskExecutor used will start tasks synchronously or
+ * asynchronously. The default setting uses a synchronous task executor.
 * 
- * There are only two required dependencies of this Launcher, a {@link JobExecutor} 
- * and a {@link JobRepository}.  The JobRepository is used to obtain a valid 
- * JobExecution.  The Repository must be used because the provided {@link Job} could
- * be a restart of an existing {@link JobInstance}, and only the Repository can 
- * reliably recreate it.  Once the execution is obtained, it can be run by
- * passing it to the JobExecutor.
+ * There is only one required dependency of this Launcher, a
+ * {@link JobRepository}. The JobRepository is used to obtain a valid
+ * JobExecution. The Repository must be used because the provided
+ * {@link JobSupport} could be a restart of an existing {@link JobInstance},
+ * and only the Repository can reliably recreate it.
 * 
 * @author Lucas Ward
 * @since 1.0
- * @see JobExecutor
 * @see JobRepository
 * @see TaskExecutor
 */
-public class SimpleJobLauncher implements JobLauncher, InitializingBean{
-	
+public class SimpleJobLauncher implements JobLauncher, InitializingBean {
+
 	protected static final Log logger = LogFactory.getLog(SimpleJobLauncher.class);

 	private JobRepository jobRepository;

-	private JobExecutor jobExecutor;
-	
 	private TaskExecutor taskExecutor = new SyncTaskExecutor();
-	
+
 	/**
-	 * Run the provided job with the given {@link JobParameters}.  The {@link JobParameters} will
-	 * be used to determine if this is an execution of an existing job instance, or if a new
-	 * one should be created.
+	 * Run the provided job with the given {@link JobParameters}. The
+	 * {@link JobParameters} will be used to determine if this is an execution
+	 * of an existing job instance, or if a new one should be created.
 	 * 
 	 * @param job, the job to be run.
-	 * @param jobParameters, the {@link JobParameters} for this particular execution.
-	 * @return JobExecutionAlreadyRunningException if the JobInstance already exists and has
-	 * an execution already running.
+	 * @param jobParameters, the {@link JobParameters} for this particular
+	 * execution.
+	 * @return JobExecutionAlreadyRunningException if the JobInstance already
+	 * exists and has an execution already running.
 	 */
 	public JobExecution run(final Job job, final JobParameters jobParameters)
-		throws JobExecutionAlreadyRunningException {
-		
+			throws JobExecutionAlreadyRunningException {
+
 		Assert.notNull(job, "The Job must not be null.");
 		Assert.notNull(jobParameters, "The JobParameters must not be null.");

 		final JobExecution jobExecution = jobRepository.createJobExecution(job, jobParameters);
-		
-		taskExecutor.execute(new Runnable(){
+
+		taskExecutor.execute(new Runnable() {

 			public void run() {
-				try{
+				try {
 					logger.info("Job: [" + job + "] launched with the following parameters: [" + jobParameters + "]");
-					ExitStatus exitStatus = jobExecutor.run(job, jobExecution);
-					//shouldn't need to set the exit status like this, I'm leaving it to make the latest change easier
+					ExitStatus exitStatus = job.run(jobExecution);
+					// shouldn't need to set the exit status like this, I'm
+					// leaving it to make the latest change easier
 					jobExecution.setExitStatus(exitStatus);
-					logger.info("Job: [" + job + "] completed successfully with the following parameters: [" 
+					logger.info("Job: [" + job + "] completed successfully with the following parameters: ["
 							+ jobParameters + "]");
 				}
-				catch(Throwable t){
-					logger.info("Job: [" + job + "] failed with the following parameters: [" 
-							+ jobParameters + "]", t);
+				catch (Throwable t) {
+					logger.info("Job: [" + job + "] failed with the following parameters: [" + jobParameters + "]", t);
 					rethrow(t);
 				}
 			}
@@ -104,8 +101,9 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean{
 					throw (RuntimeException) t;
 				}
 				throw new RuntimeException(t);
-			}});
-		
+			}
+		});
+
 		return jobExecution;
 	}

@@ -118,15 +116,6 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean{
 		this.jobRepository = jobRepository;
 	}

-	/**
-	 * Set the JobExecutor.
-	 * 
-	 * @param jobExecutor
-	 */
-	public void setJobExecutor(JobExecutor jobExecutor) {
-		this.jobExecutor = jobExecutor;
-	}
-	
 	/**
 	 * Set the TaskExecutor. (Optional)
 	 * 
@@ -136,12 +125,10 @@ public class SimpleJobLauncher implements JobLauncher, InitializingBean{
 		this.taskExecutor = taskExecutor;
 	}

-	/*
-	 * Ensure the required dependencies of a JobExecutor and 
-	 * JobRepository have been set.
+	/**
+	 * Ensure the required dependencies of a {@link JobRepository} have been set.
 	 */
 	public void afterPropertiesSet() throws Exception {
-		Assert.state(jobExecutor != null, "A JobExecutor has not been set.");
 		Assert.state(jobRepository != null, "A JobRepository has not been set.");
 		logger.info("No TaskExecutor has been set, defaulting to synchronous executor.");
 	}
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/SimpleJobRepository.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Properties;

 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -73,7 +74,7 @@ public class SimpleJobRepository implements JobRepository {
 	/**
 	 * <p>
 	 * Create a (@link {@link JobExecution}) based on the passed in
-	 * {@link JobIdentifier} and {@link Job}. However, unique identification of
+	 * {@link JobIdentifier} and {@link JobSupport}. 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 of the job is populated with the correct value.
@@ -82,7 +83,7 @@ public class SimpleJobRepository implements JobRepository {
 	 * <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
-	 * restartability. The {@link Job} restartable property will be checked
+	 * restartability. The {@link JobSupport} restartable property will be checked
 	 * first. If it is not false, a new job will be created, regardless of
 	 * whether or not one exists. If it is true, the {@link JobDao} will be
 	 * checked to determine if the job already exists, if it does, it's steps
@@ -100,9 +101,9 @@ public class SimpleJobRepository implements JobRepository {
 	 * <li>What happens then depends on how many existing job instances we
 	 * find:
 	 * <ul>
-	 * <li>If there are none, or the {@link Job} is marked restartable, then we
+	 * <li>If there are none, or the {@link JobSupport} is marked restartable, then we
 	 * create a new {@link JobInstance}</li>
-	 * <li>If there is more than one and the {@link Job} is not marked as
+	 * <li>If there is more than one and the {@link JobSupport} is not marked as
 	 * restartable, it is an error. This could be caused by a job whose
 	 * restartable flag has changed to be more strict (true not false)
 	 * <em>after</em> it has been executed at least once.</li>
@@ -124,7 +125,7 @@ public class SimpleJobRepository implements JobRepository {
 	 * platform does not support the higher isolation levels).
 	 * </p>
 	 * 
-	 * @see JobRepository#createJobExecution(Job, JobParameters)
+	 * @see JobRepository#createJobExecution(JobSupport, JobParameters)
 	 * 
 	 * @throws BatchRestartException if more than one JobInstance if found or if
 	 * JobInstance.getJobExecutionCount() is greater than Job.getStartLimit()
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapJobDao.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/repository/dao/MapJobDao.java
@@ -22,7 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;

-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -47,7 +47,7 @@ public class MapJobDao implements JobDao {

 	public JobInstance createJobInstance(String jobName, JobParameters jobParameters) {
 		JobInstance jobInstance = new JobInstance(new Long(currentId++), jobParameters);
-		jobInstance.setJob(new Job(jobName));
+		jobInstance.setJob(new JobSupport(jobName));
 		
 		jobsById.put(jobInstance.getId(), jobInstance);
 		return jobInstance;
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java
@@ -16,10 +16,13 @@
 package org.springframework.batch.execution.step.simple;

 import org.springframework.batch.core.domain.Step;
-import org.springframework.batch.core.domain.StepExecutor;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.core.domain.StepSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.tasklet.Tasklet;
+import org.springframework.batch.io.exception.BatchCriticalException;
+import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.Assert;
@@ -105,20 +108,13 @@ public abstract class AbstractStep extends StepSupport {
 		Assert.notNull(jobRepository, "JobRepository is mandatory");
 		Assert.notNull(transactionManager, "TransactionManager is mandatory");
 	}
-
-	/**
-	 * Create a {@link SimpleStepExecutor}.
-	 * 
-	 * @see org.springframework.batch.core.domain.Step#createStepExecutor()
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.domain.StepSupport#process(org.springframework.batch.core.domain.StepExecution)
 	 */
-	public StepExecutor createStepExecutor() {
-		assertMandatoryProperties();
-		SimpleStepExecutor executor = new SimpleStepExecutor(this);
-		executor.setRepository(jobRepository);
-		executor.applyConfiguration(this);
-		executor.setTasklet(tasklet);
-		executor.setTransactionManager(transactionManager);
-		return executor;
+	public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
+		SimpleStepExecutor executor = createStepExecutor();
+		return executor.process(stepExecution);
 	}

 	/**
@@ -130,4 +126,17 @@ public abstract class AbstractStep extends StepSupport {
 		this.tasklet = tasklet;
 	}

+	/**
+	 * @return
+	 */
+	protected SimpleStepExecutor createStepExecutor() {
+		assertMandatoryProperties();
+		SimpleStepExecutor executor = new SimpleStepExecutor(this);
+		executor.setRepository(jobRepository);
+		executor.applyConfiguration(this);
+		executor.setTasklet(tasklet);
+		executor.setTransactionManager(transactionManager);
+		return executor;
+	}
+
 }
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsHolder.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsHolder.java
@@ -16,16 +16,14 @@
 package org.springframework.batch.execution.step.simple;

 import org.springframework.batch.core.domain.Step;
-import org.springframework.batch.core.domain.StepExecutor;
 import org.springframework.batch.repeat.RepeatOperations;

 /**
 * Marker interface for indicating that a {@link RepeatOperations} instance is
 * available for the inner loop (chunk operations) and outer loop (step
- * operations) in a {@link StepExecutor}. The inner loop is normally going to
- * be in-process and thread-bound so it makes sense for
- * {@link Step} implementations to be able to override the
- * strategies that control that loop.
+ * operations) in a {@link Step}. The inner loop is normally going to be
+ * in-process and thread-bound so it makes sense for {@link Step}
+ * implementations to be able to override the strategies that control that loop.
 * 
 * @author Dave Syer
 * 
@@ -36,7 +34,7 @@ public interface RepeatOperationsHolder {
 	 * Principal method in the {@link RepeatOperationsHolder} interface.
 	 * 
 	 * @return a {@link RepeatOperations} which can be used to iterate over an
-	 *         inner loop (chunk).
+	 * inner loop (chunk).
 	 */
 	RepeatOperations getChunkOperations();

@@ -44,7 +42,7 @@ public interface RepeatOperationsHolder {
 	 * Additional method in the {@link RepeatOperationsHolder} interface.
 	 * 
 	 * @return a {@link RepeatOperations} which can be used to iterate over an
-	 *         outer loop (step).
+	 * outer loop (step).
 	 */
 	RepeatOperations getStepOperations();
 }
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/RepeatOperationsStep.java
@@ -17,7 +17,10 @@
 package org.springframework.batch.execution.step.simple;

 import org.springframework.batch.core.domain.Step;
-import org.springframework.batch.core.domain.StepExecutor;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInterruptedException;
+import org.springframework.batch.io.exception.BatchCriticalException;
+import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatOperations;

 /**
@@ -72,13 +75,10 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio
 		this.stepOperations = stepOperations;
 	}

-	/**
-	 * Create a new {@link SimpleStepExecutor} with the step and chunk
-	 * operations.
-	 * 
-	 * @see org.springframework.batch.core.domain.StepSupport#createStepExecutor()
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.execution.step.simple.AbstractStep#process(org.springframework.batch.core.domain.StepExecution)
 	 */
-	public StepExecutor createStepExecutor() {
+	public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
 		assertMandatoryProperties();
 		SimpleStepExecutor executor = (SimpleStepExecutor) super.createStepExecutor();
 		if (stepOperations != null) {
@@ -87,6 +87,6 @@ public class RepeatOperationsStep extends AbstractStep implements RepeatOperatio
 		if (chunkOperations != null) {
 			executor.setChunkOperations(chunkOperations);
 		}
-		return executor;
+		return executor.process(stepExecution);
 	}
 }
--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java
@@ -23,7 +23,6 @@ import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepContribution;
 import org.springframework.batch.core.domain.StepExecution;
-import org.springframework.batch.core.domain.StepExecutor;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
@@ -57,10 +56,10 @@ import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;

 /**
- * Simple implementation of {@link StepExecutor} executing the step as a set of
- * chunks, each chunk surrounded by a transaction. The structure is therefore
- * that of two nested loops, with transaction boundary around the whole inner
- * loop. The outer loop is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}),
+ * Simple implementation of executing the step as a set of chunks, each chunk
+ * surrounded by a transaction. The structure is therefore that of two nested
+ * loops, with transaction boundary around the whole inner loop. The outer loop
+ * is controlled by the step operations ({@link #setStepOperations(RepeatOperations)}),
 * and the inner loop by the chunk operations ({@link #setChunkOperations(RepeatOperations)}).
 * The inner loop should always be executed in a single thread, so the chunk
 * operations should not do any concurrent execution. N.B. usually that means
@@ -76,7 +75,7 @@ import org.springframework.util.Assert;
 * @author Lucas Ward
 * 
 */
-public class SimpleStepExecutor implements StepExecutor {
+public class SimpleStepExecutor {

 	private RepeatOperations chunkOperations = new RepeatTemplate();

--- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/StepInterruptionPolicy.java
+++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/StepInterruptionPolicy.java
@@ -16,13 +16,13 @@

 package org.springframework.batch.execution.step.simple;

-import org.springframework.batch.core.domain.StepExecutor;
+import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.repeat.RepeatContext;

 /**
 * Strategy interface for an interruption policy. This policy allows
- * {@link StepExecutor} implementations to check if a job has been interrupted.
+ * {@link Step} implementations to check if a job has been interrupted.
 * 
 * @author Lucas Ward
 * 
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/configuration/JobRegistryBeanPostProcessorTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/configuration/JobRegistryBeanPostProcessorTests.java
@@ -19,7 +19,7 @@ import java.util.Collection;

 import junit.framework.TestCase;

-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.repository.DuplicateJobException;
 import org.springframework.batch.core.repository.NoSuchJobException;
 import org.springframework.beans.FatalBeanException;
@@ -58,7 +58,7 @@ public class JobRegistryBeanPostProcessorTests extends TestCase {
 	public void testAfterInitializationWithCorrectType() throws Exception {
 		MapJobRegistry registry = new MapJobRegistry();
 		processor.setJobConfigurationRegistry(registry);
-		Job configuration = new Job();
+		JobSupport configuration = new JobSupport();
 		configuration.setBeanName("foo");
 		assertEquals(configuration, processor.postProcessAfterInitialization(
 				configuration, "bar"));
@@ -68,7 +68,7 @@ public class JobRegistryBeanPostProcessorTests extends TestCase {
 	public void testAfterInitializationWithDuplicate() throws Exception {
 		MapJobRegistry registry = new MapJobRegistry();
 		processor.setJobConfigurationRegistry(registry);
-		Job configuration = new Job();
+		JobSupport configuration = new JobSupport();
 		configuration.setBeanName("foo");
 		processor.postProcessAfterInitialization(configuration, "bar");
 		try {
@@ -83,7 +83,7 @@ public class JobRegistryBeanPostProcessorTests extends TestCase {
 	public void testUnregisterOnDestroy() throws Exception {
 		MapJobRegistry registry = new MapJobRegistry();
 		processor.setJobConfigurationRegistry(registry);
-		Job configuration = new Job();
+		JobSupport configuration = new JobSupport();
 		configuration.setBeanName("foo");
 		assertEquals(configuration, processor.postProcessAfterInitialization(
 				configuration, "bar"));
@@ -103,7 +103,7 @@ public class JobRegistryBeanPostProcessorTests extends TestCase {
 				.getBean("registry");
 		Collection configurations = registry.getJobConfigurations();
 		// System.err.println(configurations);
-		String[] names = context.getBeanNamesForType(Job.class);
+		String[] names = context.getBeanNamesForType(JobSupport.class);
 		int count = names.length;
 		// Each concrete bean of type JobConfiguration is registered...
 		assertEquals(count, configurations.size());
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/configuration/MapJobRegistryTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/configuration/MapJobRegistryTests.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import junit.framework.TestCase;

 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.repository.DuplicateJobException;
 import org.springframework.batch.core.repository.NoSuchJobException;
 import org.springframework.batch.execution.configuration.MapJobRegistry;
@@ -33,13 +34,13 @@ public class MapJobRegistryTests extends TestCase {
 	private MapJobRegistry registry = new MapJobRegistry();

 	/**
-	 * Test method for {@link org.springframework.batch.execution.configuration.MapJobRegistry#unregister(org.springframework.batch.core.domain.Job)}.
+	 * Test method for {@link org.springframework.batch.execution.configuration.MapJobRegistry#unregister(org.springframework.batch.core.domain.JobSupport)}.
 	 * @throws Exception 
 	 */
 	public void testUnregister() throws Exception {
-		registry.register(new Job("foo"));
+		registry.register(new JobSupport("foo"));
 		assertNotNull(registry.getJob("foo"));
-		registry.unregister(new Job("foo"));
+		registry.unregister(new JobSupport("foo"));
 		try {
 			assertNull(registry.getJob("foo"));
 			fail("Expected NoSuchJobConfigurationException");
@@ -54,9 +55,9 @@ public class MapJobRegistryTests extends TestCase {
 	 * Test method for {@link org.springframework.batch.execution.configuration.MapJobRegistry#getJob(java.lang.String)}.
 	 */
 	public void testReplaceDuplicateConfiguration() throws Exception {
-		registry.register(new Job("foo"));
+		registry.register(new JobSupport("foo"));
 		try {
-			registry.register(new Job("foo"));
+			registry.register(new JobSupport("foo"));
 		} catch (DuplicateJobException e) {
 			fail("Unexpected DuplicateJobConfigurationException");
 			// expected
@@ -68,7 +69,7 @@ public class MapJobRegistryTests extends TestCase {
 	 * Test method for {@link org.springframework.batch.execution.configuration.MapJobRegistry#getJob(java.lang.String)}.
 	 */
 	public void testRealDuplicateConfiguration() throws Exception {
-		Job jobConfiguration = new Job("foo");
+		Job jobConfiguration = new JobSupport("foo");
 		registry.register(jobConfiguration);
 		try {
 			registry.register(jobConfiguration);
@@ -84,9 +85,9 @@ public class MapJobRegistryTests extends TestCase {
 	 * @throws Exception 
 	 */
 	public void testGetJobConfigurations() throws Exception {
-		Job configuration = new Job("foo");
+		Job configuration = new JobSupport("foo");
 		registry.register(configuration);
-		registry.register(new Job("bar"));
+		registry.register(new JobSupport("bar"));
 		Collection configurations = registry.getJobConfigurations();
 		assertEquals(2, configurations.size());
 		assertTrue(configurations.contains(configuration));
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/DefaultJobExecutorTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/job/DefaultJobExecutorTests.java
@@ -14,7 +14,7 @@
 * limitations under the License.
 */

-package org.springframework.batch.execution.job;
+package org.springframework.batch.execution.job.simple;

 import java.util.ArrayList;
 import java.util.List;
@@ -22,13 +22,10 @@ import java.util.List;
 import junit.framework.TestCase;

 import org.springframework.batch.core.domain.BatchStatus;
-import org.springframework.batch.core.domain.Job;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
-import org.springframework.batch.core.domain.Step;
 import org.springframework.batch.core.domain.StepExecution;
-import org.springframework.batch.core.domain.StepExecutor;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
@@ -39,7 +36,6 @@ import org.springframework.batch.execution.repository.dao.JobDao;
 import org.springframework.batch.execution.repository.dao.MapJobDao;
 import org.springframework.batch.execution.repository.dao.MapStepDao;
 import org.springframework.batch.execution.repository.dao.StepDao;
-import org.springframework.batch.execution.step.simple.AbstractStep;
 import org.springframework.batch.execution.step.simple.SimpleStep;
 import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;
@@ -50,7 +46,7 @@ import org.springframework.batch.repeat.ExitStatus;
 * 
 * @author Lucas Ward
 */
-public class DefaultJobExecutorTests extends TestCase {
+public class SimpleJobTests extends TestCase {

 	private JobRepository jobRepository;

@@ -60,23 +56,7 @@ public class DefaultJobExecutorTests extends TestCase {

 	private List list = new ArrayList();

-	StepExecutor defaultStepLifecycle = new StubStepExecutor() {
-		public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException,
-				BatchCriticalException {
-			list.add("default");
-			return ExitStatus.FINISHED;
-		}
-	};
-
-	StepExecutor configurationStepLifecycle = new StubStepExecutor() {
-		public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException,
-				BatchCriticalException {
-			list.add("special");
-			return ExitStatus.FINISHED;
-		}
-	};
-
-	private JobInstance job;
+	private JobInstance jobInstance;

 	private JobExecution jobExecution;

@@ -88,15 +68,13 @@ public class DefaultJobExecutorTests extends TestCase {

 	private StepExecution stepExecution2;

-	private AbstractStep stepConfiguration1;
+	private StubStep stepConfiguration1;

-	private AbstractStep stepConfiguration2;
+	private StubStep stepConfiguration2;

-	private Job jobConfiguration;
-	
 	private JobParameters jobParameters = new JobParameters();

-	private DefaultJobExecutor jobExecutor;
+	private SimpleJob job;

 	protected void setUp() throws Exception {
 		super.setUp();
@@ -106,33 +84,34 @@ public class DefaultJobExecutorTests extends TestCase {
 		jobDao = new MapJobDao();
 		stepDao = new MapStepDao();
 		jobRepository = new SimpleJobRepository(jobDao, stepDao);
-		jobExecutor = new DefaultJobExecutor();
-		jobExecutor.setJobRepository(jobRepository);
+		job = new SimpleJob();
+		job.setJobRepository(jobRepository);

-		stepConfiguration1 = new SimpleStep("TestStep1") {
-			public StepExecutor createStepExecutor() {
-				return defaultStepLifecycle;
+		stepConfiguration1 = new StubStep("TestStep1");
+		stepConfiguration1.setCallback(new Runnable() {
+			public void run() {
+				list.add("default");
 			}
-		};
-		stepConfiguration2 = new SimpleStep("TestStep2") {
-			public StepExecutor createStepExecutor() {
-				return defaultStepLifecycle;
+		});
+		stepConfiguration2 = new StubStep("TestStep2");
+		stepConfiguration2.setCallback(new Runnable() {
+			public void run() {
+				list.add("default");
 			}
-		};
+		});
 		stepConfiguration1.setJobRepository(jobRepository);
 		stepConfiguration2.setJobRepository(jobRepository);

 		List stepConfigurations = new ArrayList();
 		stepConfigurations.add(stepConfiguration1);
 		stepConfigurations.add(stepConfiguration2);
-		jobConfiguration = new Job();
-		jobConfiguration.setName("testJob");
-		jobConfiguration.setSteps(stepConfigurations);
+		job.setName("testJob");
+		job.setSteps(stepConfigurations);

-		jobExecution = jobRepository.createJobExecution(jobConfiguration, jobParameters);
-		job = jobExecution.getJobInstance();
+		jobExecution = jobRepository.createJobExecution(job, jobParameters);
+		jobInstance = jobExecution.getJobInstance();

-		List steps = job.getStepInstances();
+		List steps = jobInstance.getStepInstances();
 		step1 = (StepInstance) steps.get(0);
 		step2 = (StepInstance) steps.get(1);
 		stepExecution1 = new StepExecution(step1, jobExecution, null);
@@ -145,18 +124,16 @@ public class DefaultJobExecutorTests extends TestCase {
 	}

 	public void testRunNormally() throws Exception {
-
 		stepConfiguration1.setStartLimit(5);
 		stepConfiguration2.setStartLimit(5);
-		jobExecutor.run(jobConfiguration, jobExecution);
+		job.run(jobExecution);
 		assertEquals(2, list.size());
 		checkRepository(BatchStatus.COMPLETED);
 	}

 	public void testRunWithSimpleStepExecutor() throws Exception {

-		jobExecutor = new DefaultJobExecutor();
-		jobExecutor.setJobRepository(jobRepository);
+		job.setJobRepository(jobRepository);
 		// do not set StepExecutorFactory...
 		stepConfiguration1.setStartLimit(5);
 		stepConfiguration1.setTasklet(new Tasklet() {
@@ -172,14 +149,15 @@ public class DefaultJobExecutorTests extends TestCase {
 				return ExitStatus.FINISHED;
 			}
 		});
-		jobExecutor.run(jobConfiguration, jobExecution);
+		job.run(jobExecution);
 		assertEquals(2, list.size());
 		checkRepository(BatchStatus.COMPLETED, ExitStatus.FINISHED);
+
 	}

 	public void testExecutionContextIsSet() throws Exception {
 		testRunNormally();
-		assertEquals(job, jobExecution.getJobInstance());
+		assertEquals(jobInstance, jobExecution.getJobInstance());
 		assertEquals(2, jobExecution.getStepExecutions().size());
 		assertEquals(step1, stepExecution1.getStep());
 		assertEquals(step2, stepExecution2.getStep());
@@ -188,42 +166,31 @@ public class DefaultJobExecutorTests extends TestCase {
 	public void testInterrupted() throws Exception {
 		stepConfiguration1.setStartLimit(5);
 		stepConfiguration2.setStartLimit(5);
-		final StepInterruptedException exception = new StepInterruptedException(
-				"Interrupt!");
-		defaultStepLifecycle = new StubStepExecutor() {
-			public ExitStatus process(StepExecution stepExecution)
-					throws StepInterruptedException, BatchCriticalException {
-				throw exception;
-			}
-		};
+		final StepInterruptedException exception = new StepInterruptedException("Interrupt!");
+		stepConfiguration1.setProcessException(exception);
 		try {
-			jobExecutor.run(jobConfiguration, jobExecution);
-		} catch (BatchCriticalException e) {
+			job.run(jobExecution);
+		}
+		catch (BatchCriticalException e) {
 			assertEquals(exception, e.getCause());
 		}
 		assertEquals(0, list.size());
-		checkRepository(BatchStatus.STOPPED, new ExitStatus(false,
-				ExitCodeExceptionClassifier.STEP_INTERRUPTED));
+		checkRepository(BatchStatus.STOPPED, new ExitStatus(false, ExitCodeExceptionClassifier.STEP_INTERRUPTED));
 	}

 	public void testFailed() throws Exception {
 		stepConfiguration1.setStartLimit(5);
 		stepConfiguration2.setStartLimit(5);
 		final RuntimeException exception = new RuntimeException("Foo!");
-		defaultStepLifecycle = new StubStepExecutor() {
-			public ExitStatus process(StepExecution stepExecution)
-					throws StepInterruptedException, BatchCriticalException {
-				throw exception;
-			}
-		};
+		stepConfiguration1.setProcessException(exception);
 		try {
-			jobExecutor.run(jobConfiguration, jobExecution);
-		} catch (RuntimeException e) {
+			job.run(jobExecution);
+		}
+		catch (RuntimeException e) {
 			assertEquals(exception, e);
 		}
 		assertEquals(0, list.size());
-		checkRepository(BatchStatus.FAILED, new ExitStatus(false,
-				ExitCodeExceptionClassifier.FATAL_EXCEPTION));
+		checkRepository(BatchStatus.FAILED, new ExitStatus(false, ExitCodeExceptionClassifier.FATAL_EXCEPTION));
 	}

 	public void testStepShouldNotStart() throws Exception {
@@ -231,64 +198,92 @@ public class DefaultJobExecutorTests extends TestCase {
 		stepConfiguration1.setStartLimit(0);

 		try {
-			jobExecutor.run(jobConfiguration, jobExecution);
+			job.run(jobExecution);
 			fail("Expected BatchCriticalException");
-		} catch (BatchCriticalException ex) {
+		}
+		catch (BatchCriticalException ex) {
 			// expected
-			assertTrue("Wrong message in exception: " + ex.getMessage(), ex
-					.getMessage().indexOf("start limit exceeded") >= 0);
+			assertTrue("Wrong message in exception: " + ex.getMessage(), ex.getMessage()
+					.indexOf("start limit exceeded") >= 0);
 		}
 	}

 	public void testNoSteps() throws Exception {
-		jobConfiguration.setSteps(new ArrayList());
+		job.setSteps(new ArrayList());

-		jobExecutor.run(jobConfiguration, jobExecution);
+		job.run(jobExecution);
 		ExitStatus exitStatus = jobExecution.getExitStatus();
-		assertTrue("Wrong message in execution: " + exitStatus, exitStatus
-				.getExitDescription().indexOf("No steps configured") >= 0);
+		assertTrue("Wrong message in execution: " + exitStatus, exitStatus.getExitDescription().indexOf(
+				"No steps configured") >= 0);
 	}

 	public void testNoStepsExecuted() throws Exception {
 		step1.setStatus(BatchStatus.COMPLETED);
 		step2.setStatus(BatchStatus.COMPLETED);

-		jobExecutor.run(jobConfiguration, jobExecution);
+		job.run(jobExecution);
 		ExitStatus exitStatus = jobExecution.getExitStatus();
-		assertTrue("Wrong message in execution: " + exitStatus, exitStatus
-				.getExitDescription().indexOf("steps already completed") >= 0);
+		assertTrue("Wrong message in execution: " + exitStatus, exitStatus.getExitDescription().indexOf(
+				"steps already completed") >= 0);
 	}

 	/*
 	 * Check JobRepository to ensure status is being saved.
 	 */
 	private void checkRepository(BatchStatus status, ExitStatus exitStatus) {
-		assertEquals(job, jobDao.findJobInstances(job.getJobName(), jobParameters).get(0));
+		assertEquals(jobInstance, jobDao.findJobInstances(jobInstance.getJobName(), jobParameters).get(0));
 		// because map dao stores in memory, it can be checked directly
-		assertEquals(status, job.getStatus());
-		JobExecution jobExecution = (JobExecution) jobDao
-				.findJobExecutions(job).get(0);
-		assertEquals(job.getId(), jobExecution.getJobId());
+		assertEquals(status, jobInstance.getStatus());
+		JobExecution jobExecution = (JobExecution) jobDao.findJobExecutions(jobInstance).get(0);
+		assertEquals(jobInstance.getId(), jobExecution.getJobId());
 		assertEquals(status, jobExecution.getStatus());
 		if (exitStatus != null) {
-			assertEquals(jobExecution.getExitStatus().getExitCode(), exitStatus
-					.getExitCode());
+			assertEquals(jobExecution.getExitStatus().getExitCode(), exitStatus.getExitCode());
 		}
 	}

 	private void checkRepository(BatchStatus status) {
 		checkRepository(status, null);
 	}
-	
-	private class StubStepExecutor implements StepExecutor {

-		public void applyConfiguration(Step configuration) {
+	private class StubStep extends SimpleStep {
+
+		private Runnable runnable;
+		private Exception exception;
+
+		/**
+		 * @param string
+		 */
+		public StubStep(String string) {
+			super(string);
 		}

-		public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException,
-				BatchCriticalException {
-			return null;
+		/**
+		 * @param exception
+		 */
+		public void setProcessException(Exception exception) {
+			this.exception = exception;
 		}
-		
+
+		/**
+		 * @param runnable
+		 */
+		public void setCallback(Runnable runnable) {
+			this.runnable = runnable;
+		}
+
+		public ExitStatus process(StepExecution stepExecution) throws StepInterruptedException, BatchCriticalException {
+			if (exception instanceof RuntimeException) {
+				throw (RuntimeException)exception;
+			}
+			if (exception instanceof StepInterruptedException) {
+				throw (StepInterruptedException)exception;
+			}
+			if (runnable!=null) {
+				runnable.run();
+			}
+			return ExitStatus.FINISHED;
+		}
+
 	}
 }
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobLauncherTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobLauncherTests.java
@@ -21,9 +21,10 @@ import junit.framework.TestCase;
 import org.easymock.MockControl;
 import org.springframework.batch.core.domain.Job;
 import org.springframework.batch.core.domain.JobExecution;
-import org.springframework.batch.core.domain.JobExecutor;
 import org.springframework.batch.core.domain.JobParameters;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.io.exception.BatchCriticalException;
 import org.springframework.batch.repeat.ExitStatus;

 /**
@@ -34,27 +35,24 @@ public class SimpleJobLauncherTests extends TestCase {

 	private SimpleJobLauncher jobLauncher;
 	
-	private JobExecutor jobExecutor;
-	private JobRepository jobRepository;
-	
-	private MockControl executorControl = MockControl.createControl(JobExecutor.class);
 	private MockControl repositoryControl = MockControl.createControl(JobRepository.class);
 	
-	private Job job = new Job("foo");
+	private Job job = new JobSupport("foo") {
+		public ExitStatus run(JobExecution execution) throws BatchCriticalException {
+			return ExitStatus.FINISHED;
+		}
+	};
 	private JobParameters jobParameters = new JobParameters();
+
+	private JobRepository jobRepository;
 	
 	protected void setUp() throws Exception {
 		super.setUp();
 		
-		jobLauncher = new SimpleJobLauncher();
-		
-		jobExecutor = (JobExecutor)executorControl.getMock();
-		jobRepository = (JobRepository)repositoryControl.getMock();
-		
-		jobLauncher.setJobExecutor(jobExecutor);
+		jobLauncher = new SimpleJobLauncher();	
+		jobRepository = (JobRepository) repositoryControl.getMock();
 		jobLauncher.setJobRepository(jobRepository);
 		
-		
 	}


@@ -64,16 +62,12 @@ public class SimpleJobLauncherTests extends TestCase {
 		
 		jobRepository.createJobExecution(job, jobParameters);
 		repositoryControl.setReturnValue(jobExecution);
-		jobExecutor.run(job, jobExecution);
-		executorControl.setDefaultReturnValue(ExitStatus.FINISHED);
 		
 		repositoryControl.replay();
-		executorControl.replay();
 		
 		jobLauncher.run(job, jobParameters);
 		assertEquals(ExitStatus.FINISHED, jobExecution.getExitStatus());
 		
 		repositoryControl.verify();
-		executorControl.verify();
 	}
 }
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java
@@ -18,17 +18,17 @@ package org.springframework.batch.execution.launch;

 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;

 import junit.framework.TestCase;

 import org.springframework.batch.core.domain.BatchStatus;
-import org.springframework.batch.core.domain.Job;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.tasklet.Tasklet;
-import org.springframework.batch.execution.job.DefaultJobExecutor;
+import org.springframework.batch.execution.job.simple.SimpleJob;
 import org.springframework.batch.execution.repository.SimpleJobRepository;
 import org.springframework.batch.execution.repository.dao.MapJobDao;
 import org.springframework.batch.execution.repository.dao.MapStepDao;
@@ -64,11 +64,11 @@ public class SimpleJobTests extends TestCase {

 	private ItemReader provider;

-	private DefaultJobExecutor jobExecutor = new DefaultJobExecutor();;
+	private SimpleJob job = new SimpleJob();;

 	protected void setUp() throws Exception {
 		super.setUp();
-		jobExecutor.setJobRepository(repository);
+		job.setJobRepository(repository);
 	}

 	private Tasklet getTasklet(String arg) throws Exception {
@@ -99,22 +99,22 @@ public class SimpleJobTests extends TestCase {

 	public void testSimpleJob() throws Exception {

-		Job jobConfiguration = new Job();
+		job.setSteps(new ArrayList());
 		AbstractStep step = new SimpleStep(getTasklet("foo", "bar"));
 		step.setJobRepository(repository);
 		step.setTransactionManager(new ResourcelessTransactionManager());
-		jobConfiguration.addStep(step);
+		job.addStep(step);
 		step = new SimpleStep(getTasklet("spam"));
 		step.setJobRepository(repository);
 		step.setTransactionManager(new ResourcelessTransactionManager());
-		jobConfiguration.addStep(step);
+		job.addStep(step);

-		JobInstance job = repository.createJobExecution(jobConfiguration, new JobParameters()).getJobInstance();
+		JobInstance jobInstance = repository.createJobExecution(job, new JobParameters()).getJobInstance();

-		JobExecution jobExecutionContext = new JobExecution(job);
+		JobExecution jobExecutionContext = new JobExecution(jobInstance);

-		jobExecutor.run(jobConfiguration, jobExecutionContext);
-		assertEquals(BatchStatus.COMPLETED, job.getStatus());
+		job.run(jobExecutionContext);
+		assertEquals(BatchStatus.COMPLETED, jobInstance.getStatus());
 		assertEquals(3, processed.size());
 		assertTrue(processed.contains("foo"));

@@ -122,7 +122,6 @@ public class SimpleJobTests extends TestCase {

 	public void testSimpleJobWithRecovery() throws Exception {

-		Job jobConfiguration = new Job();
 		final List throwables = new ArrayList();

 		RepeatTemplate chunkOperations = new RepeatTemplate();
@@ -151,10 +150,10 @@ public class SimpleJobTests extends TestCase {
 			}
 		});
 		module.afterPropertiesSet();
-		jobConfiguration.addStep(step);
+		job.setSteps(Collections.singletonList(step));

-		JobExecution jobExecution = repository.createJobExecution(jobConfiguration, new JobParameters());
-		jobExecutor.run(jobConfiguration, jobExecution);
+		JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
+		job.run(jobExecution);

 		assertEquals(BatchStatus.COMPLETED, jobExecution.getJobInstance().getStatus());
 		assertEquals(0, processed.size());
@@ -165,7 +164,6 @@ public class SimpleJobTests extends TestCase {

 	public void testExceptionTerminates() throws Exception {

-		Job jobConfiguration = new Job();
 		final ItemOrientedTasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
 		AbstractStep step = new SimpleStep(module);
 		step.setJobRepository(repository);
@@ -176,18 +174,17 @@ public class SimpleJobTests extends TestCase {
 			}
 		});
 		module.afterPropertiesSet();
-		jobConfiguration.addStep(step);
+		job.setSteps(Collections.singletonList(step));

-		JobExecution jobExecution = repository.createJobExecution(jobConfiguration, new JobParameters());
-		JobInstance job = jobExecution.getJobInstance();
+		JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
 		try {
-			jobExecutor.run(jobConfiguration, jobExecution);
+			job.run(jobExecution);
 			fail("Expected RuntimeException");
 		}
 		catch (RuntimeException e) {
 			assertEquals("Foo", e.getMessage());
 			// expected
 		}
-		assertEquals(BatchStatus.FAILED, job.getStatus());
+		assertEquals(BatchStatus.FAILED, jobExecution.getJobInstance().getStatus());
 	}
 }
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/SimpleJobRepositoryTests.java
@@ -25,7 +25,7 @@ import junit.framework.TestCase;

 import org.easymock.ArgumentsMatcher;
 import org.easymock.MockControl;
-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -51,7 +51,7 @@ public class SimpleJobRepositoryTests extends TestCase {

 	SimpleJobRepository jobRepository;

-	Job jobConfiguration;
+	JobSupport jobConfiguration;

 	JobParameters jobParameters;

@@ -89,7 +89,7 @@ public class SimpleJobRepositoryTests extends TestCase {
 		jobParameters = new JobParametersBuilder().toJobParameters();
 		

-		jobConfiguration = new Job();
+		jobConfiguration = new JobSupport();
 		jobConfiguration.setBeanName("RepositoryTest");
 		jobConfiguration.setRestartable(true);

--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractJobDaoTests.java
@@ -22,6 +22,7 @@ import java.util.Map;

 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -68,7 +69,7 @@ public abstract class AbstractJobDaoTests extends
 //		jobRuntimeInformation = new ScheduledJobIdentifier("Job1", "TestStream",
 //				new SimpleDateFormat("yyyyMMdd").parse("20070505"));

-		job = new Job("Job1");
+		job = new JobSupport("Job1");
 		
 		// Create job.
 		jobInstance = jobDao.createJobInstance(job.getName(), jobParameters);
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/repository/dao/AbstractStepDaoTests.java
@@ -22,6 +22,7 @@ import java.util.Properties;

 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -82,7 +83,7 @@ public abstract class AbstractStepDaoTests extends AbstractTransactionalDataSour
 	 * @see org.springframework.test.AbstractTransactionalSpringContextTests#onSetUpInTransaction()
 	 */
 	protected void onSetUpInTransaction() throws Exception {
-		Job job = new Job("TestJob");
+		Job job = new JobSupport("TestJob");
 		jobInstance = jobDao.createJobInstance(job.getName(), jobParameters);
 		step1 = stepDao.createStep(jobInstance, "TestStep1");
 		step2 = stepDao.createStep(jobInstance, "TestStep2");
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/resource/BatchResourceFactoryBeanTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/resource/BatchResourceFactoryBeanTests.java
@@ -21,7 +21,7 @@ import java.io.IOException;

 import junit.framework.TestCase;

-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -64,7 +64,7 @@ public class BatchResourceFactoryBeanTests extends TestCase {
 		resourceFactory.setRootDirectory(rootDir);
 		
 		jobInstance = new JobInstance(new Long(0), new JobParameters());
-		jobInstance.setJob(new Job("testJob"));
+		jobInstance.setJob(new JobSupport("testJob"));
 		JobExecution jobExecution = new JobExecution(jobInstance);
 		StepInstance step = new StepInstance(jobInstance, "bar");
 		StepExecution stepExecution = new StepExecution(step, jobExecution, null);
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/resource/DefaultJobInstanceLabelGeneratorTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/resource/DefaultJobInstanceLabelGeneratorTests.java
@@ -6,7 +6,7 @@ import java.util.Date;

 import junit.framework.TestCase;

-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.domain.JobParametersBuilder;
@@ -33,7 +33,7 @@ public class DefaultJobInstanceLabelGeneratorTests extends TestCase {
 	 * Test method for {@link org.springframework.batch.execution.resource.DefaultJobInstanceLabelGenerator#getLabel()}.
 	 */
 	public void testGetLabel() {
-		assertEquals("foo", instance.getLabel(new JobInstance(null, new JobParameters(), new Job("foo"))));
+		assertEquals("foo", instance.getLabel(new JobInstance(null, new JobParameters(), new JobSupport("foo"))));
 	}
 	
 	/**
@@ -41,7 +41,7 @@ public class DefaultJobInstanceLabelGeneratorTests extends TestCase {
 	 */
 	public void testDefaultGetLabel() throws Exception {
 		JobParameters jobParameters = new JobParametersBuilder().addDate("schedule.date", DATE).addString("key", "bar").toJobParameters();
-		JobInstance job = new JobInstance(null, jobParameters, new Job(null));
+		JobInstance job = new JobInstance(null, jobParameters, new JobSupport(null));
 		assertEquals("null-bar-"+dateFormat.format(DATE), instance.getLabel(job));
 	}

@@ -50,7 +50,7 @@ public class DefaultJobInstanceLabelGeneratorTests extends TestCase {
 	 */
 	public void testGetLabelWithAllProperties() throws Exception {
 		JobParameters jobParameters = new JobParametersBuilder().addDate("schedule.date", DATE).addString("key", "bar").toJobParameters();
-		JobInstance job = new JobInstance(null, jobParameters, new Job("foo"));
+		JobInstance job = new JobInstance(null, jobParameters, new JobSupport("foo"));
 		assertEquals("foo-bar-"+dateFormat.format(DATE), instance.getLabel(job));
 	}

--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/RepeatOperationsStepTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/RepeatOperationsStepTests.java
@@ -76,13 +76,11 @@ public class RepeatOperationsStepTests extends TestCase {
 		configuration.setChunkOperations(repeatTemplate);
 		configuration.setJobRepository(new JobRepositorySupport());
 		configuration.setTransactionManager(new ResourcelessTransactionManager());
-		SimpleStepExecutor executor = (SimpleStepExecutor) configuration
-				.createStepExecutor();
 		StepExecution stepExecution = new StepExecution(new StepInstance(
 				new Long(11)), new JobExecution(new JobInstance(new Long(0L), new JobParameters()),
 				new Long(12)));
 		try {
-			executor.process(stepExecution);
+			configuration.process(stepExecution);
 			fail("Expected RuntimeException");
 		} catch (NullPointerException e) {
 			// expected
@@ -117,12 +115,10 @@ public class RepeatOperationsStepTests extends TestCase {
 				return ExitStatus.CONTINUABLE;
 			}
 		});
-		SimpleStepExecutor executor = (SimpleStepExecutor) configuration
-				.createStepExecutor();
 		StepExecution stepExecution = new StepExecution(new StepInstance(
 				new Long(11)), new JobExecution(new JobInstance(new Long(0L), new JobParameters()),
 				new Long(12)));
-		executor.process(stepExecution);
+		configuration.process(stepExecution);
 		assertEquals(2, list.size());
 		assertEquals(1, steps.size());
 	}
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java
@@ -25,7 +25,7 @@ import java.util.Properties;

 import junit.framework.TestCase;

-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
@@ -113,7 +113,7 @@ public class SimpleStepExecutorTests extends TestCase {
 		stepExecutor.setChunkOperations(template);
 		
 		jobInstance = new JobInstance(new Long(0), new JobParameters());
-		jobInstance.setJob(new Job("FOO"));
+		jobInstance.setJob(new JobSupport("FOO"));
 	}

 	public void testStepExecutor() throws Exception {
--- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java
+++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/StepExecutorInterruptionTests.java
@@ -21,12 +21,11 @@ import java.util.List;
 import junit.framework.TestCase;

 import org.springframework.batch.core.domain.BatchStatus;
-import org.springframework.batch.core.domain.Job;
+import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.JobExecution;
 import org.springframework.batch.core.domain.JobInstance;
 import org.springframework.batch.core.domain.JobParameters;
 import org.springframework.batch.core.domain.StepExecution;
-import org.springframework.batch.core.domain.StepExecutor;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.domain.StepInterruptedException;
 import org.springframework.batch.core.repository.JobRepository;
@@ -53,13 +52,11 @@ public class StepExecutorInterruptionTests extends TestCase {

 	private RepeatOperationsStep stepConfiguration;

-	private StepExecutor executor;
-
 	public void setUp() throws Exception {

 		jobRepository = new SimpleJobRepository(jobDao, stepDao);

-		Job jobConfiguration = new Job();
+		JobSupport jobConfiguration = new JobSupport();
 		stepConfiguration = new RepeatOperationsStep();
 		jobConfiguration.addStep(stepConfiguration);
 		jobConfiguration.setBeanName("testJob");
@@ -85,12 +82,11 @@ public class StepExecutorInterruptionTests extends TestCase {
 				return new ExitStatus(foo != 1);
 			}
 		});
-		executor = stepConfiguration.createStepExecutor();

 		Thread processingThread = new Thread() {
 			public void run() {
 				try {
-					executor.process(stepExecution);
+					stepConfiguration.process(stepExecution);
 				}
 				catch (StepInterruptedException e) {
 					// do nothing...
--- a/spring-batch-execution/src/test/resources/job-configuration.xml
+++ b/spring-batch-execution/src/test/resources/job-configuration.xml
@@ -11,7 +11,7 @@
 		<property name="jobConfigurationRegistry" ref="jobRegistry"/>
 	</bean>

-	<bean id="test-job" class="org.springframework.batch.core.domain.Job">
+	<bean id="test-job" class="org.springframework.batch.core.domain.JobSupport">
 		<property name="steps">
 			<bean id="step1" class="org.springframework.batch.execution.step.simple.SimpleStep">
 				<constructor-arg>
--- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml
+++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/bootstrap/support/job.xml
@@ -7,7 +7,7 @@
 		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
 		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd">

-	<bean id="test-job" class="org.springframework.batch.core.domain.Job">
+	<bean id="test-job" class="org.springframework.batch.core.domain.JobSupport">
 		<property name="steps">
 			<bean id="step1" class="org.springframework.batch.execution.step.simple.SimpleStep">
 				<constructor-arg>
--- a/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml
+++ b/spring-batch-execution/src/test/resources/org/springframework/batch/execution/configuration/test-context.xml
@@ -13,7 +13,7 @@
 	
 	<bean id="registry" class="org.springframework.batch.execution.configuration.MapJobRegistry"/>

-	<bean id="test-job" class="org.springframework.batch.core.domain.Job">
+	<bean id="test-job" class="org.springframework.batch.core.domain.JobSupport">
 		<property name="steps">
 			<bean id="step1" class="org.springframework.batch.execution.step.simple.SimpleStep">
 				<constructor-arg>
@@ -34,15 +34,15 @@
 		</property>
 	</bean>
 	
-	<bean id="test-job-with-name" class="org.springframework.batch.core.domain.Job">
+	<bean id="test-job-with-name" class="org.springframework.batch.core.domain.JobSupport">
 		<property name="name" value="foo"/>
 	</bean>

-	<bean id="test-job-with-bean-name" class="org.springframework.batch.core.domain.Job">
+	<bean id="test-job-with-bean-name" class="org.springframework.batch.core.domain.JobSupport">
 		<property name="beanName" value="bar"/>
 	</bean>

-	<bean id="abstract-job" class="org.springframework.batch.core.domain.Job" abstract="true"/>
+	<bean id="abstract-job" class="org.springframework.batch.core.domain.JobSupport" abstract="true"/>
 	
 	<bean id="test-job-with-parent" parent="abstract-job"/>

@@ -50,7 +50,7 @@

 	<bean id="test-job-with-parent-and-bean-name" parent="abstract-job" p:beanName="bucket"/>

-	<bean id="parent-job" class="org.springframework.batch.core.domain.Job"/>
+	<bean id="parent-job" class="org.springframework.batch.core.domain.JobSupport"/>
 	
 	<bean id="test-job-with-concrete-parent" parent="parent-job" p:name="maps"/>

--- a/spring-batch-execution/src/test/resources/simple-container-definition.xml
+++ b/spring-batch-execution/src/test/resources/simple-container-definition.xml
@@ -12,20 +12,15 @@
 	<bean id="simpleContainerLauncher"
 		class="org.springframework.batch.execution.launch.SimpleJobLauncher">
 		<property name="jobRepository" ref="simpleJobRepository" />
-		<property name="jobExecutor" ref="jobLifecycle" />
 	</bean>

 	<bean id="jobRegistry"
 		class="org.springframework.batch.execution.configuration.MapJobRegistry" />

-	<bean id="jobLifecycle"
-		class="org.springframework.batch.execution.job.DefaultJobExecutor">
-		<property name="jobRepository" ref="simpleJobRepository" />
-	</bean>
-
 	<bean id="simpleJob"
-		class="org.springframework.batch.core.domain.Job" abstract="true">
+		class="org.springframework.batch.execution.job.simple.SimpleJob" abstract="true">
 		<property name="restartable" value="true" />
+		<property name="jobRepository" ref="simpleJobRepository" />
 	</bean>

 	<bean id="simpleStep"