diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
index 0a89448a8..90bf9b53c 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
@@ -59,6 +59,8 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 
 	private CompletionPolicy chunkCompletionPolicy;
 
+	private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT;
+
 	/**
 	 * Set the commit interval. Either set this or the chunkCompletionPolicy but
 	 * not both.
@@ -125,6 +127,17 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 		this.taskExecutor = taskExecutor;
 	}
 
+	/**
+	 * Public setter for the throttle limit. This limits the number of tasks
+	 * queued for concurrent processing to prevent thread pools from being
+	 * overwhelmed. Defaults to
+	 * {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
+	 * @param throttleLimit the throttle limit to set.
+	 */
+	public void setThrottleLimit(int throttleLimit) {
+		this.throttleLimit = throttleLimit;
+	}
+
 	/**
 	 * Public getter for the ItemHandler.
 	 * @return the ItemHandler
@@ -159,9 +172,10 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 		if (taskExecutor != null) {
 			TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
 			repeatTemplate.setTaskExecutor(taskExecutor);
+			repeatTemplate.setThrottleLimit(throttleLimit);
 			stepOperations = repeatTemplate;
 		}
-		
+
 		stepOperations.setExceptionHandler(exceptionHandler);
 
 		step.setStepOperations(stepOperations);
@@ -175,8 +189,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 	private CompletionPolicy getChunkCompletionPolicy() {
 		Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0),
 				"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
-		Assert.state(commitInterval >= 0,
-				"The commitInterval must be positive or zero (for default value).");
+		Assert.state(commitInterval >= 0, "The commitInterval must be positive or zero (for default value).");
 
 		if (chunkCompletionPolicy != null) {
 			return chunkCompletionPolicy;
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
index 720f8e82f..8ea9c10c0 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
@@ -47,6 +47,7 @@ import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
 /**
  * Tests for {@link SimpleStepFactoryBean}.
@@ -123,6 +124,25 @@ public class SimpleStepFactoryBeanTests extends TestCase {
 		assertTrue(written.contains("foo"));
 	}
 
+	public void testSimpleConcurrentJob() throws Exception {
+
+		job.setSteps(new ArrayList());
+		SimpleStepFactoryBean factory = getStepFactory("foo", "bar");
+		factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
+		factory.setThrottleLimit(1);
+
+		AbstractStep step = (AbstractStep) factory.getObject();
+		step.setName("step1");
+		job.addStep(step);
+
+		JobExecution jobExecution = repository.createJobExecution(job, new JobParameters());
+
+		job.execute(jobExecution);
+		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
+		assertEquals(2, written.size());
+		assertTrue(written.contains("foo"));
+	}
+
 	public void testSimpleJobWithItemListeners() throws Exception {
 
 		final List throwables = new ArrayList();