From b6aca5fa05fff81a34d4e24f3faa3e4576249e45 Mon Sep 17 00:00:00 2001
From: dsyer <dsyer@2e8e966d-6415-0410-94b9-dc4e65435d03>
Date: Mon, 1 Mar 2010 10:21:57 +0000
Subject: [PATCH] RESOLVED - issue BATCH-1518: Add optimization for
 commit-interval=1

---
 .../item/FaultTolerantChunkProcessor.java     | 10 ++++--
 .../FaultTolerantChunkProcessorTests.java     | 32 ++++++++++++++++---
 ...aultTolerantStepFactoryBeanRetryTests.java |  8 ++---
 3 files changed, 40 insertions(+), 10 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 a85deeea7..8585a114f 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
@@ -208,7 +208,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 					O output = null;
 					try {
 						count.incrementAndGet();
-						O cached = (cacheIterator != null&& cacheIterator.hasNext()) ? cacheIterator.next() : null;
+						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
@@ -221,7 +221,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 						}
 						else {
 							output = doProcess(item);
-							if (!processorTransactional) { 
+							if (!processorTransactional) {
 								cache.add(output);
 							}
 						}
@@ -377,6 +377,12 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 								"Retry exhausted after last attempt in recovery path, but exception is not skippable.",
 								context.getLastThrowable());
 					}
+					
+					if (inputs.size()==1 && outputs.getSkips().isEmpty()) {
+						// Optimization for trivial chunk size: no need to scan
+						checkSkipPolicy(inputs.iterator(), outputs.iterator(), context.getLastThrowable(), contribution);
+						return null;
+					}
 
 					inputs.setBusy(true);
 					scan(contribution, inputs, outputs, chunkMonitor);
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 faad7430a..bd1900bb9 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
@@ -154,6 +154,30 @@ public class FaultTolerantChunkProcessorTests {
 		assertEquals(0, contribution.getFilterCount());
 	}
 
+	@Test
+	public void testWriteSkipOnExceptionWithTrivialChunk() 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("fail"));
+		try {
+			processor.process(contribution, inputs);
+			fail("Expected RuntimeException");
+		}
+		catch (RuntimeException e) {
+			assertEquals("Expected Exception!", e.getMessage());
+		}
+		processor.process(contribution, inputs);
+		assertEquals(1, contribution.getSkipCount());
+		assertEquals(0, contribution.getWriteCount());
+		assertEquals(0, contribution.getFilterCount());
+	}
+
 	@Test
 	public void testTransformWithExceptionAndNoRollback() throws Exception {
 		processor.setItemProcessor(new ItemProcessor<String, String>() {
@@ -200,11 +224,11 @@ public class FaultTolerantChunkProcessorTests {
 		assertEquals(1, chunk.getItems().size());
 		processor.process(contribution, chunk);
 		assertEquals(0, chunk.getItems().size());
-		// foo is written twice because the failure is detected on the second
-		// attempt when throttling
+		// foo is written once because it the failure is detected before it is
+		// committed the first time
 		assertEquals("[foo, bar]", list.toString());
-		// but the after listener is only called once, which is important
-		assertEquals(2, after.size());
+		// the after listener is called once per successful item, which is important
+		assertEquals("[foo, bar]", after.toString());
 	}
 
 	@Test
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
index 936aa2a82..8a17ed88c 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
@@ -304,7 +304,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(expectedOutput, written);
 
 		assertEquals("[a, b, c, d, e, f, null]", provided.toString());
-		assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", processed.toString());
+		assertEquals("[a, b, b, b, b, b, c, d, d, d, d, d, e, f]", processed.toString());
 		assertEquals("[b, d]", recovered.toString());
 	}
 
@@ -361,7 +361,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(7, provided.size());
 		// [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, d, e, f, d,
 		// e, f, d, e, f, d, e, f, d, e, f, d, e, f]
-		System.err.println(processed);
+		// System.err.println(processed);
 		assertEquals(36, processed.size());
 		// [b, d]
 		assertEquals(2, recovered.size());
@@ -405,7 +405,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(1, provided.size());
 		// the failed items are tried up to the limit (but only precisely so if
 		// the commit interval is 1)
-		assertEquals("[b, b, b, b, b]", processed.toString());
+		assertEquals("[b, b, b, b]", processed.toString());
 		// []
 		assertEquals(0, recovered.size());
 		assertEquals(1, stepExecution.getReadCount());
@@ -496,7 +496,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(0, stepExecution.getSkipCount());
 		// [b]
 		assertEquals(1, provided.size());
-		assertEquals("[b, b, b, b, b]", processed.toString());
+		assertEquals("[b, b, b, b]", processed.toString());
 		// []
 		assertEquals(0, recovered.size());
 		assertEquals(1, stepExecution.getReadCount());