diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
index 786dc03c8..878a3444a 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
@@ -43,6 +43,7 @@ import org.springframework.batch.retry.policy.MapRetryContextCache;
 import org.springframework.batch.retry.policy.RetryContextCache;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
 import org.springframework.batch.support.Classifier;
+import org.springframework.util.Assert;
 
 /**
  * Factory bean for step that provides options for configuring skip behaviour.
@@ -78,7 +79,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private int cacheCapacity = 0;
 
-	private int retryLimit = 0;
+	private int retryLimit = 1;
 
 	private Collection<Class<? extends Throwable>> retryableExceptionClasses = new HashSet<Class<? extends Throwable>>();
 
@@ -109,10 +110,13 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	/**
 	 * Public setter for the retry limit. Each item can be retried up to this
-	 * limit.
-	 * @param retryLimit the retry limit to set
+	 * limit. Note this limit includes the initial attempt to process the item,
+	 * therefore <code>retryLimit == 1</code> by default.
+	 * 
+	 * @param retryLimit the retry limit to set, must be greater or equal to 1.
 	 */
 	public void setRetryLimit(int retryLimit) {
+		Assert.isTrue(retryLimit >= 1, "retry limit must be greater or equal to 1");
 		this.retryLimit = retryLimit;
 	}
 
@@ -211,7 +215,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 	protected void applyConfiguration(TaskletStep step) {
 		super.applyConfiguration(step);
 
-		if (retryLimit > 0 || skipLimit > 0 || retryPolicy != null) {
+		if (retryLimit > 1 || skipLimit > 0 || retryPolicy != null) {
 
 			addFatalExceptionIfMissing(SkipLimitExceededException.class);
 			addFatalExceptionIfMissing(NonSkippableReadException.class);
@@ -233,7 +237,10 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 					exceptionTypeMap.put(cls, simpleRetryPolicy);
 				}
 				classifierRetryPolicy.setPolicyMap(exceptionTypeMap);
-				retryPolicy = classifierRetryPolicy;
+
+				// TODO use the classifier wrapper above to take care of fatal
+				// exceptions, regardless of the injected retry policy
+				retryPolicy = simpleRetryPolicy;
 
 			}
 			BatchRetryTemplate batchRetryTemplate = new BatchRetryTemplate();
@@ -270,7 +277,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 			exceptions.addAll(new ArrayList<Class<? extends Throwable>>(retryableExceptionClasses));
 			SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
 					new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
-			
+
 			Classifier<Throwable, Boolean> rollbackClassifier = new Classifier<Throwable, Boolean>() {
 				public Boolean classify(Throwable classifiable) {
 					return getTransactionAttribute().rollbackOn(classifiable);
@@ -280,17 +287,23 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 			FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
 					getChunkOperations());
 			chunkProvider.setSkipPolicy(readSkipPolicy);
-			chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>>getListeners(getListeners(), ItemReadListener.class));
-			chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T,S>>getListeners(getListeners(), SkipListener.class));
+			chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(getListeners(),
+					ItemReadListener.class));
+			chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
+					SkipListener.class));
 
-			FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(), getItemWriter(), batchRetryTemplate);
+			FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(
+					getItemProcessor(), getItemWriter(), batchRetryTemplate);
 			chunkProcessor.setBuffering(!isReaderTransactionalQueue);
 			chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
 			chunkProcessor.setProcessSkipPolicy(writeSkipPolicy);
 			chunkProcessor.setRollbackClassifier(rollbackClassifier);
-			chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T,S>>getListeners(getListeners(), ItemProcessListener.class));
-			chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>>getListeners(getListeners(), ItemWriteListener.class));
-			chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T,S>>getListeners(getListeners(), SkipListener.class));
+			chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(
+					getListeners(), ItemProcessListener.class));
+			chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(getListeners(),
+					ItemWriteListener.class));
+			chunkProcessor.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
+					SkipListener.class));
 
 			ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
 			tasklet.setBuffering(!isReaderTransactionalQueue);
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
index 802f34cc8..ab50b5ade 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/SimpleRetryPolicy.java
@@ -48,7 +48,7 @@ public class SimpleRetryPolicy implements RetryPolicy {
 	 */
 	public final static int DEFAULT_MAX_ATTEMPTS = 3;
 
-	private volatile int maxAttempts;
+	private volatile int maxAttempts = 1;
 
 	private BinaryExceptionClassifier retryableClassifier = new BinaryExceptionClassifier();
 
@@ -66,7 +66,7 @@ public class SimpleRetryPolicy implements RetryPolicy {
 	 * Create a {@link SimpleRetryPolicy} with the specified number of retry
 	 * attempts, and default exceptions to retry.
 	 * 
-	 * @param maxAttempts
+	 * @param maxAttempts number of allowed attempts (typically >= 1)
 	 */
 	public SimpleRetryPolicy(int maxAttempts) {
 		super();
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
index f2de38676..df5ccf0d2 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/SimpleRetryPolicyTests.java
@@ -16,36 +16,40 @@
 
 package org.springframework.batch.retry.policy;
 
+import static org.junit.Assert.*;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 
-import junit.framework.TestCase;
 
+import org.junit.Test;
 import org.springframework.batch.retry.RetryContext;
 
-public class SimpleRetryPolicyTests extends TestCase {
+public class SimpleRetryPolicyTests {
 
+	@Test
 	public void testCanRetryIfNoException() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);
 		assertTrue(policy.canRetry(context));
 	}
 
-	@SuppressWarnings("unchecked")
+	@Test
 	public void testEmptyExceptionsNeverRetry() throws Exception {
 
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);
 
 		// We can't retry any exceptions...
-		policy.setRetryableExceptionClasses(Collections.EMPTY_SET);
+		Collection<Class<? extends Throwable>> empty = Collections.emptySet();
+		policy.setRetryableExceptionClasses(empty);
 
 		// ...so we can't retry this one...
 		policy.registerThrowable(context, new IllegalStateException());
 		assertFalse(policy.canRetry(context));
 	}
 
+	@Test
 	public void testRetryLimitInitialState() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);
@@ -55,6 +59,7 @@ public class SimpleRetryPolicyTests extends TestCase {
 		assertFalse(policy.canRetry(context));
 	}
 
+	@Test
 	public void testRetryLimitSubsequentState() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);
@@ -66,6 +71,7 @@ public class SimpleRetryPolicyTests extends TestCase {
 		assertFalse(policy.canRetry(context));
 	}
 
+	@Test
 	public void testRetryCount() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);
@@ -77,6 +83,7 @@ public class SimpleRetryPolicyTests extends TestCase {
 		assertEquals("foo", context.getLastThrowable().getMessage());
 	}
 
+	@Test
 	public void testFatalOverridesRetryable() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		policy.setFatalExceptionClasses(getClasses(Exception.class));
@@ -97,6 +104,7 @@ public class SimpleRetryPolicyTests extends TestCase {
 		return classes;
 	}
 
+	@Test
 	public void testParent() throws Exception {
 		SimpleRetryPolicy policy = new SimpleRetryPolicy();
 		RetryContext context = policy.open(null);