diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/ChunkElementParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/ChunkElementParser.java
index 8280c5b63..8e9c41aa6 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/ChunkElementParser.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/ChunkElementParser.java
@@ -118,6 +118,11 @@ public class ChunkElementParser {
 			propertyValues.addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue);
 		}
 
+		String isProcessorTransactional = element.getAttribute("processor-transactional");
+		if (StringUtils.hasText(isProcessorTransactional)) {
+			propertyValues.addPropertyValue("processorTransactional", isProcessorTransactional);
+		}
+
 		handleExceptionElement(element, parserContext, propertyValues, "skippable-exception-classes",
 				"skippableExceptionClasses");
 
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
index 485133930..35b32ebda 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
@@ -102,7 +102,9 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
 
 	private Integer commitInterval;
 
-	private Boolean isReaderTransactionalQueue;
+	private Boolean readerTransactionalQueue;
+
+	private Boolean processorTransactional;
 
 	private Integer retryLimit;
 
@@ -228,8 +230,11 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
 		if (cacheCapacity != null) {
 			fb.setCacheCapacity(cacheCapacity);
 		}
-		if (isReaderTransactionalQueue != null) {
-			fb.setIsReaderTransactionalQueue(isReaderTransactionalQueue);
+		if (readerTransactionalQueue != null) {
+			fb.setIsReaderTransactionalQueue(readerTransactionalQueue);
+		}
+		if (processorTransactional != null) {
+			fb.setProcessorTransactional(processorTransactional);
 		}
 		if (retryLimit != null) {
 			fb.setRetryLimit(retryLimit);
@@ -307,7 +312,25 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
 	private void validateFaultTolerantSettings() {
 		validateDependency("skippable-exception-classes", skippableExceptionClasses, "skip-limit", skipLimit, true);
 		validateDependency("retryable-exception-classes", retryableExceptionClasses, "retry-limit", retryLimit, true);
+		validateAtLeastOneDependency("processor-transactional", processorTransactional, "'retry-limit' or 'skip-limit'", retryLimit, skipLimit);
 		validateDependency("retry-listeners", retryListeners, "retry-limit", retryLimit, false);
+		if (isPresent(processorTransactional) && !processorTransactional && isPresent(readerTransactionalQueue) && readerTransactionalQueue) {
+			throw new IllegalArgumentException("The field 'processor-transactional' cannot be false if 'reader-transactional-queue' is true");
+		}
+	}
+
+	private void validateAtLeastOneDependency(String dependantName, Boolean dependantValue, String name, Object... values) {
+		boolean oneIsPresent = false;
+		for (Object value : values) {
+			if (isPresent(value)) {
+				oneIsPresent = true;
+				break;
+			}
+		}
+		if (isPresent(dependantValue) && !oneIsPresent) {
+			throw new IllegalArgumentException("The field '" + dependantName + "' is not permitted on the step ["
+					+ this.name + "] because " + name + " is not present.");
+		}
 	}
 
 	private void validateDependency(String dependantName, Object dependantValue, String name, Object value,
@@ -331,7 +354,7 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
 
 	private boolean isFaultTolerant() {
 		return isPositive(skipLimit) || isPositive(retryLimit) || isPositive(cacheCapacity)
-				|| isTrue(isReaderTransactionalQueue);
+				|| isTrue(readerTransactionalQueue);
 	}
 
 	private boolean isTrue(Boolean b) {
@@ -527,7 +550,19 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
 	 * @param isReaderTransactionalQueue the value of the flag
 	 */
 	public void setIsReaderTransactionalQueue(boolean isReaderTransactionalQueue) {
-		this.isReaderTransactionalQueue = isReaderTransactionalQueue;
+		this.readerTransactionalQueue = isReaderTransactionalQueue;
+	}
+
+	/**
+	 * Flag to signal that the processor is transactional, in which case it
+	 * should be called for every item in every transaction. If false then we
+	 * can cache the processor results between transactions in the case of a
+	 * rollback.
+	 * 
+	 * @param processorTransactional the value to set
+	 */
+	public void setProcessorTransactional(Boolean processorTransactional) {
+		this.processorTransactional = processorTransactional;
 	}
 
 	/**
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 423e2bbf5..6b85fe771 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
@@ -18,6 +18,7 @@ package org.springframework.batch.core.step.item;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -62,6 +63,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 	private ChunkMonitor chunkMonitor = new ChunkMonitor();
 
+	private boolean processorTransactional;
+
 	/**
 	 * The {@link KeyGenerator} to use to identify failed items across rollback.
 	 * Not used in the case of the {@link #setBuffering(boolean) buffering flag}
@@ -107,7 +110,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 	/**
 	 * A flag to indicate that items have been buffered and therefore will
 	 * always come back as a chunk after a rollback. Otherwise things are more
-	 * complicated because after a rollback the new chunk might or moght not
+	 * complicated because after a rollback the new chunk might or might not
 	 * contain items from the previous failed chunk.
 	 * 
 	 * @param buffering
@@ -116,6 +119,13 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		this.buffering = buffering;
 	}
 
+	/**
+	 * @param processorTransactional
+	 */
+	public void setProcessorTransactional(boolean processorTransactional) {
+		this.processorTransactional = processorTransactional;
+	}
+
 	public FaultTolerantChunkProcessor(ItemProcessor<? super I, ? extends O> itemProcessor,
 			ItemWriter<? super O> itemWriter, BatchRetryTemplate batchRetryTemplate) {
 		super(itemProcessor, itemWriter);
@@ -183,12 +193,13 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		Chunk<O> outputs = new Chunk<O>();
 		@SuppressWarnings("unchecked")
 		UserData<O> data = (UserData<O>) inputs.getUserData();
-		Chunk<O> cache = data.getOutputs();
-		final Chunk<O>.ChunkIterator cacheIterator = cache.isEmpty() ? null : cache.iterator();
+		final Chunk<O> cache = data.getOutputs();
+		final Iterator<O> cacheIterator = cache.isEmpty() ? null : new ArrayList<O>(cache.getItems()).iterator();
 		final AtomicInteger count = new AtomicInteger(0);
 
 		for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
 
+			final int scanLimit = processorTransactional ? 1 : 0;
 			final I item = iterator.next();
 
 			RetryCallback<O> retryCallback = new RetryCallback<O>() {
@@ -197,8 +208,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					O output = null;
 					try {
 						count.incrementAndGet();
-						O cached = (cacheIterator != null) ? cacheIterator.next() : null;
-						if (cached != null && count.get() > 1) {
+						O cached = (cacheIterator != null&& cacheIterator.hasNext()) ? cacheIterator.next() : null;
+						if (cached != null && count.get() > scanLimit) {
 							/*
 							 * If there is a cached chunk then we must be
 							 * scanning for errors in the writer, in which case
@@ -210,6 +221,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 						}
 						else {
 							output = doProcess(item);
+							if (!processorTransactional) { 
+								cache.add(output);
+							}
 						}
 					}
 					catch (Exception e) {
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 061b328a3..d56d2911f 100755
--- 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
@@ -100,6 +100,8 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 
 	private ChunkMonitor chunkMonitor = new ChunkMonitor();
 
+	private boolean processorTransactional= true;
+
 	/**
 	 * The {@link KeyGenerator} to use to identify failed items across rollback.
 	 * Not used in the case of the
@@ -233,6 +235,13 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		this.noRollbackExceptionClasses = noRollbackExceptionClasses;
 	}
 
+	/**
+	 * @param processorTransactional
+	 */
+	public void setProcessorTransactional(boolean processorTransactional) {
+		this.processorTransactional = processorTransactional;
+	}
+
 	/**
 	 * Convenience method for subclasses to get an exception classifier based on
 	 * the provided transaction attributes.
@@ -362,6 +371,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
 		FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(getItemProcessor(),
 				getItemWriter(), batchRetryTemplate);
 		chunkProcessor.setBuffering(!isReaderTransactionalQueue());
+		chunkProcessor.setProcessorTransactional(processorTransactional);
 
 		SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses());
 		chunkProcessor.setWriteSkipPolicy(writeSkipPolicy);
diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd
index 99c682b53..b4d607cda 100644
--- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd
+++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd
@@ -744,6 +744,16 @@
 				]]></xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>
+		<xsd:attribute name="processor-transactional" type="xsd:string"
+			use="optional">
+			<xsd:annotation>
+				<xsd:documentation><![CDATA[
+				Whether the processor is transaction aware. If it is then processed items should not be
+				cached in between transactions in case of a rollback.  N.B. if reader-transactional-queue
+				is true then so should this be.
+				]]></xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="chunk-completion-policy" type="xsd:string"
 			use="optional">
 			<xsd:annotation>
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
index 894711432..c8279cbf9 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
@@ -17,7 +17,9 @@ package org.springframework.batch.core.configuration.xml;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -25,11 +27,13 @@ import java.util.Map;
 
 import org.junit.Test;
 import org.springframework.batch.core.Step;
+import org.springframework.batch.core.step.item.SimpleChunkProcessor;
 import org.springframework.batch.core.step.tasklet.TaskletStep;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.support.CompositeItemStream;
 import org.springframework.batch.retry.RetryListener;
 import org.springframework.batch.retry.listener.RetryListenerSupport;
+import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -39,17 +43,66 @@ import org.springframework.test.util.ReflectionTestUtils;
 
 /**
  * @author Dan Garrette
+ * @author Dave Syer
  * @since 2.0
  */
 public class ChunkElementParserTests {
 
-	private ConfigurableApplicationContext chunkElementParentAttributeParserTestsContext = new ClassPathXmlApplicationContext(
-			"org/springframework/batch/core/configuration/xml/ChunkElementParentAttributeParserTests-context.xml");
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testSimpleAttributes() throws Exception {
+		ConfigurableApplicationContext chunkElementAttributeParserTestsContext = new ClassPathXmlApplicationContext(
+				"org/springframework/batch/core/configuration/xml/ChunkElementSimpleAttributeParserTests-context.xml");
+		Object step = chunkElementAttributeParserTestsContext.getBean("s1", Step.class);
+		assertNotNull("Step not parsed", step);
+		Object tasklet = ReflectionTestUtils.getField(step, "tasklet");
+		Object chunkProcessor = ReflectionTestUtils.getField(tasklet, "chunkProcessor");
+		assertTrue("Wrong processor type", chunkProcessor instanceof SimpleChunkProcessor);
+	}
+
+	@Test
+	public void testProcessorTransactionalAttributes() throws Exception {
+		ConfigurableApplicationContext chunkElementAttributeParserTestsContext = new ClassPathXmlApplicationContext(
+				"org/springframework/batch/core/configuration/xml/ChunkElementTransactionalAttributeParserTests-context.xml");
+		Object step = chunkElementAttributeParserTestsContext.getBean("s1", Step.class);
+		assertNotNull("Step not parsed", step);
+		Object tasklet = ReflectionTestUtils.getField(step, "tasklet");
+		Object chunkProcessor = ReflectionTestUtils.getField(tasklet, "chunkProcessor");
+		Boolean processorTransactional = (Boolean) ReflectionTestUtils.getField(chunkProcessor,
+				"processorTransactional");
+		assertFalse("Flag not set", processorTransactional);
+	}
+
+	@Test
+	public void testProcessorTransactionalNotAllowedOnSimpleProcessor() throws Exception {
+		try {
+			new ClassPathXmlApplicationContext(
+					"org/springframework/batch/core/configuration/xml/ChunkElementIllegalAttributeParserTests-context.xml");
+			fail("Expected BeanCreationException");
+		}
+		catch (BeanCreationException e) {
+			String msg = e.getMessage();
+			assertTrue("Wrong message: " +msg, msg.contains("The field 'processor-transactional' is not permitted"));
+		}
+	}
+
+	@Test
+	public void testProcessorNonTransactionalNotAllowedWithTransactionalReader() throws Exception {
+		try {
+			new ClassPathXmlApplicationContext(
+					"org/springframework/batch/core/configuration/xml/ChunkElementIllegalTransactionalAttributeParserTests-context.xml");
+			fail("Expected BeanCreationException");
+		}
+		catch (BeanCreationException e) {
+			String msg = e.getMessage();
+			assertTrue("Wrong message: " +msg, msg.contains("The field 'processor-transactional' cannot be false if 'reader-transactional"));
+		}
+
+	}
 
 	@Test
 	public void testInheritSkippable() throws Exception {
-		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s1",
-				chunkElementParentAttributeParserTestsContext);
+		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s1", getContext());
 		assertEquals(11, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		containsClassified(skippable, ArithmeticException.class, true);
@@ -59,8 +112,7 @@ public class ChunkElementParserTests {
 
 	@Test
 	public void testInheritSkippableWithNoMerge() throws Exception {
-		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s2",
-				chunkElementParentAttributeParserTestsContext);
+		Map<Class<? extends Throwable>, Boolean> skippable = getExceptionClasses("s2", getContext());
 		assertEquals(9, skippable.size());
 		containsClassified(skippable, NullPointerException.class, true);
 		assertFalse(skippable.containsKey(ArithmeticException.class));
@@ -68,15 +120,9 @@ public class ChunkElementParserTests {
 		assertFalse(skippable.containsKey(DeadlockLoserDataAccessException.class));
 	}
 
-	private void containsClassified(Map<Class<? extends Throwable>, Boolean> classified,
-			Class<? extends Throwable> cls, boolean include) {
-		assertTrue(classified.containsKey(cls));
-		assertEquals(include, classified.get(cls));
-	}
-
 	@Test
 	public void testInheritStreams() throws Exception {
-		Collection<ItemStream> streams = getStreams("s1", chunkElementParentAttributeParserTestsContext);
+		Collection<ItemStream> streams = getStreams("s1", getContext());
 		assertEquals(2, streams.size());
 		boolean c = false;
 		for (ItemStream o : streams) {
@@ -89,8 +135,7 @@ public class ChunkElementParserTests {
 
 	@Test
 	public void testInheritRetryListeners() throws Exception {
-		Collection<RetryListener> retryListeners = getRetryListeners("s1",
-				chunkElementParentAttributeParserTestsContext);
+		Collection<RetryListener> retryListeners = getRetryListeners("s1", getContext());
 		assertEquals(2, retryListeners.size());
 		boolean g = false;
 		boolean h = false;
@@ -108,7 +153,7 @@ public class ChunkElementParserTests {
 
 	@Test
 	public void testInheritStreamsWithNoMerge() throws Exception {
-		Collection<ItemStream> streams = getStreams("s2", chunkElementParentAttributeParserTestsContext);
+		Collection<ItemStream> streams = getStreams("s2", getContext());
 		assertEquals(1, streams.size());
 		boolean c = false;
 		for (ItemStream o : streams) {
@@ -121,8 +166,7 @@ public class ChunkElementParserTests {
 
 	@Test
 	public void testInheritRetryListenersWithNoMerge() throws Exception {
-		Collection<RetryListener> retryListeners = getRetryListeners("s2",
-				chunkElementParentAttributeParserTestsContext);
+		Collection<RetryListener> retryListeners = getRetryListeners("s2", getContext());
 		assertEquals(1, retryListeners.size());
 		boolean h = false;
 		for (RetryListener o : retryListeners) {
@@ -133,6 +177,12 @@ public class ChunkElementParserTests {
 		assertTrue(h);
 	}
 
+	private void containsClassified(Map<Class<? extends Throwable>, Boolean> classified,
+			Class<? extends Throwable> cls, boolean include) {
+		assertTrue(classified.containsKey(cls));
+		assertEquals(include, classified.get(cls));
+	}
+
 	@SuppressWarnings("unchecked")
 	private Map<Class<? extends Throwable>, Boolean> getExceptionClasses(String stepName, ApplicationContext ctx)
 			throws Exception {
@@ -173,4 +223,12 @@ public class ChunkElementParserTests {
 		RetryListener[] listeners = (RetryListener[]) ReflectionTestUtils.getField(regular, "listeners");
 		return Arrays.asList(listeners);
 	}
+
+	/**
+	 * @return the chunkElementParentAttributeParserTestsContext
+	 */
+	private ConfigurableApplicationContext getContext() {
+		return new ClassPathXmlApplicationContext(
+				"org/springframework/batch/core/configuration/xml/ChunkElementParentAttributeParserTests-context.xml");
+	}
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyItemProcessor.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyItemProcessor.java
new file mode 100644
index 000000000..74e7b6ebe
--- /dev/null
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/DummyItemProcessor.java
@@ -0,0 +1,15 @@
+package org.springframework.batch.core.configuration.xml;
+
+import org.springframework.batch.item.ItemProcessor;
+
+/**
+ * @author Dave Syer
+ * @since 2.1
+ */
+public class DummyItemProcessor implements ItemProcessor<Object,Object> {
+
+	public Object process(Object item) throws Exception {
+		return item;
+	}
+
+}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java
index 548c7573b..8e2321313 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java
@@ -88,7 +88,7 @@ public class StepWithFaultTolerantProcessTaskJobParserTests {
 		assertEquals("wrong transaction-attribute:", Isolation.DEFAULT, ReflectionTestUtils.getField(factory,
 				"isolation"));
 		assertEquals("wrong transaction-attribute:", 10, ReflectionTestUtils.getField(factory, "transactionTimeout"));
-		Object txq = ReflectionTestUtils.getField(factory, "isReaderTransactionalQueue");
+		Object txq = ReflectionTestUtils.getField(factory, "readerTransactionalQueue");
 		assertEquals("wrong reader-transactional-queue:", true, txq);
 		Object te = ReflectionTestUtils.getField(factory, "taskExecutor");
 		assertEquals("wrong task-executor:", ConcurrentTaskExecutor.class, te.getClass());
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
index a7e4a59b1..8c4a767e2 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java
@@ -275,6 +275,7 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 	 */
 	@Test
 	public void testWriterNoRollbackOnRuntimeException() throws Exception {
+
 		writer.setFailures("2", "3");
 		writer.setExceptionType(SkippableRuntimeException.class);
 
@@ -340,10 +341,25 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		assertEquals("[1, 3, 5]", processor.getCommitted().toString());
 		assertEquals("[1, 3, 5]", writer.getWritten().toString());
 		assertEquals("[1, 3, 5]", writer.getCommitted().toString());
-		// TODO: Fix this with BATCH-1259?
 		assertEquals("[1, 2, 1, 3, 4, 1, 3, 5]", processor.getProcessed().toString());
 	}
 
+	@Test
+	public void testMultipleSkipsInNonTransactionalProcessor() throws Exception {
+		processor.setFailures("2", "4");
+		factory.setCommitInterval(30);
+		factory.setProcessorTransactional(false);
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+
+		assertEquals("[1, 3, 5]", writer.getWritten().toString());
+		assertEquals("[1, 3, 5]", writer.getCommitted().toString());
+		assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
+	}
+
 	@Test
 	public void testFilterInProcessor() throws Exception {
 		processor.setFailures("4");
@@ -374,10 +390,25 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		assertEquals("[1, 2, 3, 5]", processor.getCommitted().toString());
 		assertEquals("[1, 2, 3, 5]", writer.getCommitted().toString());
 		assertEquals("[1, 2, 3, 4, 1, 2, 3, 4, 5]", writer.getWritten().toString());
-		// TODO: Fix this with BATCH-1259?
 		assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5]", processor.getProcessed().toString());
 	}
 
+	@Test
+	public void testSkipInWriterNonTransactionalProcessor() throws Exception {
+		writer.setFailures("4");
+		factory.setCommitInterval(30);
+		factory.setProcessorTransactional(false);
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+
+		assertEquals("[1, 2, 3, 5]", writer.getCommitted().toString());
+		assertEquals("[1, 2, 3, 4, 1, 2, 3, 4, 5]", writer.getWritten().toString());
+		assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
+	}
+
 	@Test
 	public void testMultithreadedSkipInWriter() throws Exception {
 		writer.setFailures("1", "2", "3", "4", "5");
@@ -408,10 +439,25 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 		assertEquals("[1, 3, 5]", writer.getCommitted().toString());
 		assertEquals("[1, 2, 1, 2, 3, 4, 5]", writer.getWritten().toString());
 		assertEquals("[1, 3, 5]", processor.getCommitted().toString());
-		// TODO: Fix this with BATCH-1259?
 		assertEquals("[1, 2, 3, 4, 5, 1, 2, 3, 4, 5]", processor.getProcessed().toString());
 	}
 
+	@Test
+	public void testMultipleSkipsInWriterNonTransactionalProcessor() throws Exception {
+		writer.setFailures("2", "4");
+		factory.setCommitInterval(30);
+		factory.setProcessorTransactional(false);
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+
+		assertEquals("[1, 3, 5]", writer.getCommitted().toString());
+		assertEquals("[1, 2, 1, 2, 3, 4, 5]", writer.getWritten().toString());
+		assertEquals("[1, 2, 3, 4, 5]", processor.getProcessed().toString());
+	}
+
 	@SuppressWarnings("unchecked")
 	private Collection<Class<? extends Throwable>> getExceptionList(Class<? extends Throwable> arg) {
 		return Arrays.<Class<? extends Throwable>> asList(arg);
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalAttributeParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalAttributeParserTests-context.xml
new file mode 100644
index 000000000..ffffee634
--- /dev/null
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalAttributeParserTests-context.xml
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
+	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
+
+	<beans:import resource="common-context.xml" />
+
+	<job id="job">
+		<step id="s1">
+			<tasklet>
+				<chunk reader="reader" writer="writer" processor="processor" commit-interval="5" processor-transactional="false" />
+			</tasklet>
+		</step>
+	</job>
+
+</beans:beans>
\ No newline at end of file
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalTransactionalAttributeParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalTransactionalAttributeParserTests-context.xml
new file mode 100644
index 000000000..7228a235e
--- /dev/null
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementIllegalTransactionalAttributeParserTests-context.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
+	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
+
+	<beans:import resource="common-context.xml" />
+
+	<job id="job">
+		<step id="s1">
+			<tasklet>
+				<chunk reader="reader" writer="writer" processor="processor" skip-limit="3" commit-interval="5"
+					processor-transactional="false" reader-transactional-queue="true">
+					<skippable-exception-classes>
+						<include class="org.springframework.batch.item.validator.ValidationException" />
+					</skippable-exception-classes>
+				</chunk>
+			</tasklet>
+		</step>
+	</job>
+
+</beans:beans>
\ No newline at end of file
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementSimpleAttributeParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementSimpleAttributeParserTests-context.xml
new file mode 100644
index 000000000..8b204ef84
--- /dev/null
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementSimpleAttributeParserTests-context.xml
@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
+	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
+
+	<beans:import resource="common-context.xml" />
+
+	<job id="job">
+		<step id="s1">
+			<tasklet>
+				<chunk reader="reader" writer="writer" processor="processor" commit-interval="5"/>
+			</tasklet>
+		</step>
+	</job>
+
+</beans:beans>
\ No newline at end of file
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementTransactionalAttributeParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementTransactionalAttributeParserTests-context.xml
new file mode 100644
index 000000000..bf09e036b
--- /dev/null
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/ChunkElementTransactionalAttributeParserTests-context.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans:beans xmlns="http://www.springframework.org/schema/batch" xmlns:beans="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
+	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
+
+	<beans:import resource="common-context.xml" />
+
+	<job id="job">
+		<step id="s1">
+			<tasklet>
+				<chunk reader="reader" writer="writer" processor="processor" commit-interval="5" skip-limit="3"
+					processor-transactional="false">
+					<skippable-exception-classes>
+						<include class="org.springframework.batch.item.validator.ValidationException" />
+					</skippable-exception-classes>
+				</chunk>
+			</tasklet>
+		</step>
+	</job>
+
+</beans:beans>
\ No newline at end of file
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml
index b95eddb02..dc774d895 100644
--- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/common-context.xml
@@ -37,6 +37,7 @@
 
 	<bean id="reader" class="org.springframework.batch.core.configuration.xml.DummyItemReader"/>
 	<bean id="writer" class="org.springframework.batch.core.configuration.xml.DummyItemWriter"/>
+	<bean id="processor" class="org.springframework.batch.core.configuration.xml.DummyItemProcessor"/>
 
 	<bean id="dummyTasklet" class="org.springframework.batch.core.configuration.xml.DummyTasklet"/>