diff --git a/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactory.java b/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactory.java
new file mode 100644
index 000000000..e4d3e327a
--- /dev/null
+++ b/execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactory.java
@@ -0,0 +1,99 @@
+/*
+ * Copyright 2006-2007 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.execution.step.simple;
+
+import org.springframework.batch.core.configuration.StepConfiguration;
+import org.springframework.batch.core.executor.StepExecutor;
+import org.springframework.batch.core.executor.StepExecutorFactory;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.execution.step.SimpleStepConfiguration;
+import org.springframework.batch.repeat.RepeatOperations;
+import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
+import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
+import org.springframework.batch.repeat.support.RepeatTemplate;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link StepExecutorFactory} that only knows how to create
+ * {@link SimpleStepExecutor} instances.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class SimpleStepExecutorFactory implements StepExecutorFactory,
+		InitializingBean {
+
+	private JobRepository jobRepository;
+
+	/**
+	 * Create a {@link SimpleStepExecutor} for this configuration. If the
+	 * configuration is a {@link SimpleStepConfiguration} then a
+	 * {@link StepExecutor} is created with policies matching the commit
+	 * interval of the configuration. <br/>
+	 * 
+	 * @throws IllegalStateException
+	 *             if the configuration is not a {@link SimpleStepConfiguration}.
+	 * @throws IllegalStateException
+	 *             if the {@link JobRepository} is null.
+	 * 
+	 * @see StepExecutorFactory#getExecutor(StepConfiguration)
+	 */
+	public StepExecutor getExecutor(StepConfiguration configuration) {
+
+		Assert.notNull(jobRepository, "JobRepository cannot be null");
+		Assert.state(configuration instanceof SimpleStepConfiguration,
+				"StepConfiguration must be instance of SimpleStepConfiguration - found: ["
+						+ (configuration == null ? null : configuration
+								.getClass()) + "]");
+
+		SimpleStepExecutor executor = new SimpleStepExecutor();
+		executor.setRepository(jobRepository);
+		RepeatTemplate template = new RepeatTemplate();
+		RepeatOperations repeatOperations = template;
+		SimpleStepConfiguration simpleConfiguration = (SimpleStepConfiguration) configuration;
+		template.setCompletionPolicy(new SimpleCompletionPolicy(
+				simpleConfiguration.getCommitInterval()));
+		ExceptionHandler exceptionHandler = simpleConfiguration.getExceptionHandler();
+		if (exceptionHandler!=null) {
+			template.setExceptionHandler(exceptionHandler);
+		}
+
+		executor.setChunkOperations(repeatOperations);
+
+		return executor;
+
+	}
+
+	/**
+	 * Public setter for {@link JobRepository}.
+	 * 
+	 * @param jobRepository
+	 *            is a mandatory dependence (no default).
+	 */
+	public void setJobRepository(JobRepository jobRepository) {
+		this.jobRepository = jobRepository;
+	}
+
+	/**
+	 * Assert that all mandatory properties are set (the {@link JobRepository}).
+	 * 
+	 * @throws Exception
+	 */
+	public void afterPropertiesSet() throws Exception {
+		Assert.notNull(jobRepository);
+	}
+}
diff --git a/execution/src/test/java/org/springframework/batch/execution/step/SimpleStepExecutorFactoryTests.java b/execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactoryTests.java
similarity index 59%
rename from execution/src/test/java/org/springframework/batch/execution/step/SimpleStepExecutorFactoryTests.java
rename to execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactoryTests.java
index 09f621138..745a44244 100644
--- a/execution/src/test/java/org/springframework/batch/execution/step/SimpleStepExecutorFactoryTests.java
+++ b/execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorFactoryTests.java
@@ -13,12 +13,22 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.batch.execution.step;
+package org.springframework.batch.execution.step.simple;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 
 import junit.framework.TestCase;
 
 import org.springframework.batch.core.configuration.StepConfigurationSupport;
-import org.springframework.batch.execution.step.simple.JobRepositorySupport;
+import org.springframework.batch.core.domain.JobExecution;
+import org.springframework.batch.core.domain.JobInstance;
+import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.batch.execution.step.SimpleStepConfiguration;
+import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
 
 /**
  * @author Dave Syer
@@ -36,6 +46,27 @@ public class SimpleStepExecutorFactoryTests extends TestCase {
 		assertNotNull(factory.getExecutor(new SimpleStepConfiguration()));
 	}
 
+	public void testSuccessfulExceptionHandler() throws Exception {
+		SimpleStepConfiguration configuration = new SimpleStepConfiguration();
+		final List list = new ArrayList();
+		configuration.setExceptionHandler(new ExceptionHandler() {
+			public void handleExceptions(RepeatContext context,
+					Collection throwables) throws RuntimeException {
+				list.addAll(throwables);
+				throw new RuntimeException("Oops");
+			}
+		});
+		SimpleStepExecutor executor = (SimpleStepExecutor) factory.getExecutor(configuration);
+		StepExecution stepExecution = new StepExecution(new StepInstance(new Long(11)), new JobExecution(new JobInstance(null), new Long(12)));
+		try {
+			executor.processChunk(configuration, stepExecution);
+			fail("Expected RuntimeException");
+		} catch (RuntimeException e) {
+			assertEquals("Oops", e.getMessage());
+		}
+		assertEquals(1, list.size());
+	}
+
 	public void testUnsuccessfulWrongConfiguration() throws Exception {
 		try {
 			factory.getExecutor(new StepConfigurationSupport());