From 0c892da257dd98a3b1c9a3c2830a543ab0c878ad Mon Sep 17 00:00:00 2001
From: jpraet <jimmy.praet@telenet.be>
Date: Sat, 15 Jun 2013 21:31:38 +0200
Subject: [PATCH] BATCH-2036: Output incorrect when using
 processor-transactional="false" and skips.

---
 .../item/FaultTolerantChunkProcessor.java     |  7 +-
 .../FaultTolerantChunkProcessorTests.java     | 74 +++++++++++++++++++
 2 files changed, 78 insertions(+), 3 deletions(-)

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 664db55a9..c3956ac49 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
@@ -223,7 +223,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 						}
 						else {
 							output = doProcess(item);
-							if (!processorTransactional && !data.scanning()) {
+							if (output == null) {
+								data.incrementFilterCount();
+							} else if (!processorTransactional && !data.scanning()) {
 								cache.add(output);
 							}
 						}
@@ -254,8 +256,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					}
 					if (output == null) {
 						// No need to re-process filtered items
-						iterator.remove();
-						data.incrementFilterCount();
+						iterator.remove();						
 					}
 					return output;
 				}
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 e94bcbf72..b4e7d7b66 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
@@ -486,6 +486,80 @@ public class FaultTolerantChunkProcessorTests {
 		assertEquals(1, contribution.getWriteCount());
 		assertEquals(0, contribution.getFilterCount());
 	}
+	
+	@Test
+	// BATCH-2036
+	public void testProcessFilterAndSkippableException() throws Exception {
+		final List<String> processedItems = new ArrayList<String>();
+		processor.setProcessorTransactional(false);
+		processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemProcessor(new ItemProcessor<String, String>() {
+			@Override
+			public String process(String item) throws Exception {
+				processedItems.add(item);
+				if (item.contains("fail")) {
+					throw new IllegalArgumentException("Expected Skippable Exception!");
+				}
+				if (item.contains("skip")) {
+					return null;
+				}
+				return item;
+			}
+		});
+		processor.afterPropertiesSet();
+		Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2", "skip", "skip", "3", "fail", "fail", "4", "5"));
+		try {
+			processor.process(contribution, inputs);	
+			fail("Expected IllegalArgumentException");
+		} catch (IllegalArgumentException e) {			
+			assertEquals("Expected Skippable Exception!", e.getMessage());
+		}
+		try {
+			processor.process(contribution, inputs);	
+			fail("Expected IllegalArgumentException");
+		} catch (IllegalArgumentException e) {			
+			assertEquals("Expected Skippable Exception!", e.getMessage());
+		}
+		processor.process(contribution, inputs);			
+		assertEquals(5, list.size());
+		assertEquals("[1, 2, 3, 4, 5]", list.toString());
+		assertEquals(2, contribution.getFilterCount());
+		assertEquals(2, contribution.getProcessSkipCount());
+		assertEquals(9, processedItems.size());
+		assertEquals("[1, 2, skip, skip, 3, fail, fail, 4, 5]", processedItems.toString());
+	}
+
+	@Test
+	// BATCH-2036
+	public void testProcessFilterAndSkippableExceptionNoRollback() throws Exception {
+		final List<String> processedItems = new ArrayList<String>();
+		processor.setProcessorTransactional(false);
+		processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());
+		processor.setItemProcessor(new ItemProcessor<String, String>() {
+			@Override
+			public String process(String item) throws Exception {
+				processedItems.add(item);
+				if (item.contains("fail")) {
+					throw new IllegalArgumentException("Expected Skippable Exception!");
+				}
+				if (item.contains("skip")) {
+					return null;
+				}
+				return item;
+			}
+		});
+		processor.setRollbackClassifier(new BinaryExceptionClassifier(Collections
+				.<Class<? extends Throwable>> singleton(IllegalArgumentException.class), false));
+		processor.afterPropertiesSet();
+		Chunk<String> inputs = new Chunk<String>(Arrays.asList("1", "2", "skip", "skip", "3", "fail", "fail", "4", "5"));
+		processor.process(contribution, inputs);
+		assertEquals(5, list.size());
+		assertEquals("[1, 2, 3, 4, 5]", list.toString());
+		assertEquals(2, contribution.getFilterCount());
+		assertEquals(2, contribution.getProcessSkipCount());
+		assertEquals(9, processedItems.size());
+		assertEquals("[1, 2, skip, skip, 3, fail, fail, 4, 5]", processedItems.toString());
+	}
 
 	protected void processAndExpectPlannedRuntimeException(Chunk<String> chunk)
 			throws Exception {