diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
index 6330fb532..a8d6aab10 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
@@ -283,8 +283,11 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					try {
 						doWrite(outputs.getItems());
 					}
-					catch (Exception e) {
-						if (rollbackClassifier.classify(e)) {
+					catch (Throwable t) {
+						Exception e = (t instanceof Exception) ? (Exception) t : new IllegalStateException(
+								"Unexpected non-Exception Throwable", t);
+
+						if (rollbackClassifier.classify(t)) {
 							throw e;
 						}
 						/*
@@ -312,7 +315,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 				public Object recover(RetryContext context) throws Exception {
 
-					Exception e = (Exception) context.getLastThrowable();
+					Exception e = context.getLastThrowable();
 					if (outputs.size() > 1 && !rollbackClassifier.classify(e)) {
 						throw new RetryException("Invalid retry state during write caused by "
 								+ "exception that does not classify for rollback: ", e);
@@ -384,7 +387,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			if (item == null) {
 				continue;
 			}
-			Exception e = wrapper.getException();
+			Throwable e = wrapper.getException();
 			try {
 				getListener().onSkipInProcess(item, e);
 			}
@@ -394,7 +397,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		}
 
 		for (SkipWrapper<O> wrapper : outputs.getSkips()) {
-			Exception e = wrapper.getException();
+			Throwable e = wrapper.getException();
 			try {
 				getListener().onSkipInWrite(wrapper.getItem(), e);
 			}
@@ -464,16 +467,20 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 			inputIterator.remove();
 			outputIterator.remove();
 		}
-		catch (Exception e) {
-			if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(e)) {
+		catch (Throwable t) {
+
+			Exception e = (t instanceof Exception) ? (Exception) t : new IllegalStateException(
+					"Unexpected non-Exception Throwable", t);
+
+			if (!itemWriteSkipPolicy.shouldSkip(e, -1) && !rollbackClassifier.classify(t)) {
 				inputIterator.remove();
 				outputIterator.remove();
 			}
 			else {
 				checkSkipPolicy(inputIterator, outputIterator, e, contribution);
 			}
-			if (rollbackClassifier.classify(e)) {
-				throw e;
+			if (rollbackClassifier.classify(t)) {
+			throw e;
 			}
 		}
 		chunkMonitor.incrementOffset();
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
index 001785826..a7962adc0 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java
@@ -1,6 +1,7 @@
 package org.springframework.batch.core.step.item;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 
 import java.util.ArrayList;
@@ -87,7 +88,8 @@ public class FaultTolerantChunkProcessorTests {
 		try {
 			processor.process(contribution, inputs);
 			fail("Expected Exception");
-		} catch (Exception e) {
+		}
+		catch (Exception e) {
 			assertEquals("Skippable", e.getMessage());
 		}
 		processor.process(contribution, inputs);
@@ -96,16 +98,78 @@ public class FaultTolerantChunkProcessorTests {
 		assertEquals(1, contribution.getFilterCount());
 	}
 
+	@Test
+	public void testWriteSkipOnError() throws Exception {
+		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (items.contains("fail")) {
+					assertFalse("Expected Error!", true);
+				}
+			}
+		});
+		Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected Error");
+		}
+		catch (IllegalStateException e) {
+			assertEquals("Expected Error!", e.getCause().getMessage());
+		}
+		processor.process(contribution, inputs);
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected Error");
+		}
+		catch (IllegalStateException e) {
+			assertEquals("Expected Error!", e.getCause().getMessage());
+		}
+		assertEquals(1, contribution.getSkipCount());
+		assertEquals(1, contribution.getWriteCount());
+	}
+
+	@Test
+	public void testWriteSkipOnException() throws Exception {
+		processor.setWriteSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (items.contains("fail")) {
+					throw new RuntimeException("Expected Exception!");
+				}
+			}
+		});
+		Chunk<String> inputs = new Chunk<String>(Arrays.asList("3", "fail", "2"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		processor.process(contribution, inputs);
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		assertEquals(1, contribution.getSkipCount());
+		assertEquals(1, contribution.getWriteCount());
+	}
+
 	@Test
 	public void testTransformWithExceptionAndNoRollback() throws Exception {
 		processor.setItemProcessor(new ItemProcessor<String, String>() {
 			public String process(String item) throws Exception {
-				if (item.equals("1")) throw new DataIntegrityViolationException("Planned");
+				if (item.equals("1"))
+					throw new DataIntegrityViolationException("Planned");
 				return item;
 			}
 		});
 		processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
-		processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class), false));
+		processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections
+				.<Class<? extends Throwable>> singleton(DataIntegrityViolationException.class), false));
 		Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2"));
 		processor.process(contribution, inputs);
 		assertEquals(1, list.size());