diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
index 922c3e7c4..f589a18c6 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
@@ -32,6 +32,8 @@ public class StepContribution {
 
 	private int skipCount;
 
+	private int readSkipCount;
+
 	/**
 	 * @param execution
 	 */
@@ -72,10 +74,11 @@ public class StepContribution {
 
 	/**
 	 * @return the sum of skips accumulated in the parent {@link StepExecution}
-	 * and this <code>StepContribution</code>.
+	 * and this <code>StepContribution</code>, including uncommitted read
+	 * skips.
 	 */
 	public int getStepSkipCount() {
-		return skipCount + parentSkipCount;
+		return readSkipCount + skipCount + parentSkipCount;
 	}
 
 	/**
@@ -88,18 +91,25 @@ public class StepContribution {
 	}
 
 	/**
-	 * Increment the skip count
+	 * Increment the total skip count for this contribution
 	 */
 	public void incrementSkipCount() {
 		skipCount++;
 	}
 
 	/**
-	 * Increment the skip count by a non-trivial amount
-	 * @param delta
+	 * Increment the counter for skipped reads
 	 */
-	public void incrementSkipCount(int delta) {
-		skipCount += delta;
+	public void incrementReadSkipCount() {
+		this.readSkipCount++;
+	}
+
+	/**
+	 * Clear the read skips and transfer them to the total skip count
+	 */
+	public void commitReadSkipCount() {
+		skipCount += readSkipCount;
+		readSkipCount = 0;
 	}
 
 	/*
@@ -107,7 +117,8 @@ public class StepContribution {
 	 * @see java.lang.Object#toString()
 	 */
 	public String toString() {
-		return "[StepContribution: items="+itemCount+", commits=" + commitCount + ", skips=" + skipCount + "]";
+		return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", readSkips=" + readSkipCount
+				+ ", skips=" + skipCount + "]";
 	}
 
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
index 0190f5f91..acfce1a38 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
@@ -299,6 +299,7 @@ public class ItemOrientedStep extends AbstractStep {
 							Thread.currentThread().interrupt();
 						}
 
+						contribution.commitReadSkipCount();
 						// Apply the contribution to the step
 						// only if chunk was successful
 						stepExecution.apply(contribution);
@@ -330,13 +331,11 @@ public class ItemOrientedStep extends AbstractStep {
 
 					}
 					catch (Error e) {
-						stepExecution.incrementSkipCountBy(contribution.getSkipCount());
-						processRollback(stepExecution, fatalException, transaction);
+						processRollback(stepExecution, contribution, fatalException, transaction);
 						throw e;
 					}
 					catch (Exception e) {
-						stepExecution.incrementSkipCountBy(contribution.getSkipCount());
-						processRollback(stepExecution, fatalException, transaction);
+						processRollback(stepExecution, contribution, fatalException, transaction);
 						throw e;
 					}
 					finally {
@@ -525,11 +524,14 @@ public class ItemOrientedStep extends AbstractStep {
 
 	/**
 	 * @param stepExecution
+	 * @param contribution 
 	 * @param fatalException
 	 * @param transaction
 	 */
-	private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException,
+	private void processRollback(final StepExecution stepExecution, final StepContribution contribution, final ExceptionHolder fatalException,
 			TransactionStatus transaction) {
+
+		stepExecution.incrementSkipCountBy(contribution.getSkipCount());
 		/*
 		 * Any exception thrown within the transaction should automatically
 		 * cause the transaction to rollback.
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
index fc46d4305..d4b0cf1db 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
@@ -28,6 +28,7 @@ import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.listener.CompositeSkipListener;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
+import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.item.ItemKeyGenerator;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
@@ -48,9 +49,10 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
 public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 
 	/**
-	 * Key for transaction resource that holds skipped keys until they can be removed
+	 * Key for transaction resource that holds skipped keys until they can be
+	 * removed
 	 */
-	private static final String TO_BE_REMOVED = ItemSkipPolicyItemHandler.class.getName()+".TO_BE_REMOVED";
+	private static final String TO_BE_REMOVED = ItemSkipPolicyItemHandler.class.getName() + ".TO_BE_REMOVED";
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -153,17 +155,26 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 
 			}
 			catch (Exception e) {
-				if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
-					// increment skip count and try again
-					contribution.incrementSkipCount();
-					if (listener != null) {
-						listener.onSkipInRead(e);
+				try {
+					if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
+						// increment skip count and try again
+						contribution.incrementReadSkipCount();
+						if (listener != null) {
+							listener.onSkipInRead(e);
+						}
+						logger.debug("Skipping failed input", e);
+					}
+					else {
+						// re-throw only when the skip policy runs out of
+						// patience
+						throw e;
 					}
-					logger.debug("Skipping failed input", e);
 				}
-				else {
-					// re-throw only when the skip policy runs out of patience
-					throw e;
+				catch (SkipLimitExceededException ex) {
+					// we are headed for a abnormal ending so bake in the skip
+					// count
+					contribution.commitReadSkipCount();
+					throw ex;
 				}
 			}
 
@@ -198,7 +209,7 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 			throw e;
 		}
 	}
-	
+
 	public void mark() throws MarkFailedException {
 		super.mark();
 		clearSkippedExceptions();
@@ -235,9 +246,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 			return;
 		}
 		synchronized (skippedExceptions) {
-			for (Iterator iterator = ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).iterator(); iterator.hasNext();) {
+			for (Iterator iterator = ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).iterator(); iterator
+					.hasNext();) {
 				Object key = (Object) iterator.next();
-				skippedExceptions.remove(key);				
+				skippedExceptions.remove(key);
 			}
 			TransactionSynchronizationManager.unbindResource(TO_BE_REMOVED);
 		}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
index 91f7b16a9..43cb5a394 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
@@ -140,10 +140,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
 
 		reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
 				.commaDelimitedListToSet("2,3,5"));
-		
+
 		factory.setSkipLimit(3);
 		factory.setItemReader(reader);
-		factory.setSkippableExceptionClasses(new Class[] {Exception.class});
+		factory.setSkippableExceptionClasses(new Class[] { Exception.class });
 
 		ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
 
@@ -161,8 +161,62 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
 		// writer did not skip "2" as it never made it to writer, only "4" did
 		assertTrue(reader.processed.contains("4"));
 
-		// failure on "4" tripped the skip limit so we never got to "5"
-		List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1"));
+		// failure on "4" tripped the skip limit so we never write anything
+		// ("1" was written but rolled back)
+		List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
+		assertEquals(expectedOutput, writer.written);
+
+	}
+
+	/**
+	 * Check items causing errors are skipped as expected.
+	 */
+	public void testSkipOnReadNotDoubleCounted() throws Exception {
+
+		reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
+				.commaDelimitedListToSet("2,3,5"));
+
+		factory.setSkipLimit(4);
+		factory.setItemReader(reader);
+
+		ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
+
+		StepExecution stepExecution = jobExecution.createStepExecution(step);
+
+		step.execute(stepExecution);
+		assertEquals(4, stepExecution.getSkipCount());
+
+		// skipped 2,3,4,5
+		List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
+		assertEquals(expectedOutput, writer.written);
+
+	}
+
+	/**
+	 * Check items causing errors are skipped as expected.
+	 */
+	public void testSkipOnWriteNotDoubleCounted() throws Exception {
+
+		reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6,7"), StringUtils
+				.commaDelimitedListToSet("2,3"));
+
+		writer = new SkipWriterStub(StringUtils.commaDelimitedListToSet("4,5"));
+
+		factory.setSkipLimit(4);
+		factory.setItemReader(reader);
+		factory.setItemWriter(writer);
+
+		ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
+
+		StepExecution stepExecution = jobExecution.createStepExecution(step);
+
+		step.execute(stepExecution);
+		System.err.println(writer.written);
+		System.err.println(reader.processed);
+		assertEquals(4, stepExecution.getSkipCount());
+
+		// skipped 2,3,4,5
+		List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
 		assertEquals(expectedOutput, writer.written);
 
 	}
@@ -223,6 +277,19 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
 
 		int flushIndex = -1;
 
+		private final Collection failures;
+
+		public SkipWriterStub() {
+			this(StringUtils.commaDelimitedListToSet("4"));
+		}
+
+		/**
+		 * @param commaDelimitedListToSet
+		 */
+		public SkipWriterStub(Collection failures) {
+			this.failures = failures;
+		}
+
 		public void clear() throws ClearFailedException {
 			for (int i = flushIndex + 1; i < written.size(); i++) {
 				written.remove(i);
@@ -234,10 +301,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
 		}
 
 		public void write(Object item) throws Exception {
-			written.add(item);
-			if (item.equals("4")) {
+			if (failures.contains(item)) {
 				throw new SkippableRuntimeException("exception in writer");
 			}
+			written.add(item);
 		}
 
 	}