BATCH-220: More refinements to chunking, both the chunker and dechunker now return a 'result'. The chunker returns a ChunkingResult, and the Dechunker returns a DechunkingResult, allowing for the step to handle exceptions in a uniform way.

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStep.java

@@ -21,8 +21,10 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.core.domain.BatchStatus;
 import org.springframework.batch.core.domain.Chunk;
-import org.springframework.batch.core.domain.ChunkResult;
+import org.springframework.batch.core.domain.ChunkingResult;
 import org.springframework.batch.core.domain.Dechunker;
+import org.springframework.batch.core.domain.DechunkingResult;
+import org.springframework.batch.core.domain.ItemFailureLog;
 import org.springframework.batch.core.domain.JobInterruptedException;
 import org.springframework.batch.core.domain.StepContribution;
 import org.springframework.batch.core.domain.StepExecution;
@@ -81,6 +83,8 @@ public class ChunkedStep extends AbstractStep {
 
 	// default to checking current thread for interruption.
 	private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy();
+	
+	private ItemFailureLog failureLog = new DefaultItemFailureLog();
 
 	private StreamManager streamManager;
 
@@ -107,6 +111,10 @@ public class ChunkedStep extends AbstractStep {
 	public void setStreamManager(StreamManager streamManager) {
 		this.streamManager = streamManager;
 	}
+	
+	public void setFailureLog(ItemFailureLog failureLog) {
+		this.failureLog = failureLog;
+	}
 
 	/**
 	 * Injected strategy for storage and retrieval of persistent step information. Mandatory property.
@@ -222,10 +230,14 @@ public class ChunkedStep extends AbstractStep {
 					
 					//shouldn't have to create a chunker each time, I'll refactor the interface later
 					Chunker chunker = new ItemChunker(itemReader, stepExecution);
-					final Chunk chunk = chunker.chunk(chunkSize);
-					if(chunk == null){
+					ChunkingResult chunkingResult = chunker.chunk(chunkSize);
+					
+					if(chunkingResult == null){
 						return ExitStatus.FINISHED;
 					}
+					
+					final Chunk chunk = chunkingResult.getChunk();
+					failureLog.log(chunkingResult.getExceptions());
 
 					retryTemplate.execute(new RetryCallback(){
 
@@ -292,7 +304,8 @@ public class ChunkedStep extends AbstractStep {
 
 			Dechunker dechunker = new ItemDechunker(itemWriter, stepExecution);
 			
-			ChunkResult chunkResult = dechunker.dechunk(chunk);
+			DechunkingResult chunkResult = dechunker.dechunk(chunk);
+			failureLog.log(chunkResult.getExceptions());
 
 			// TODO: check that stepExecution can
 			// aggregate these contributions if they

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java

@@ -16,6 +16,7 @@
 package org.springframework.batch.execution.step.simple;
 
 import org.springframework.batch.core.domain.Chunk;
+import org.springframework.batch.core.domain.ChunkingResult;
 import org.springframework.batch.io.exception.ReadFailureException;
 
 
@@ -36,6 +37,6 @@ public interface Chunker {
 	 * @return the {@link Chunk} that has been read.
 	 * @throws IllegalArgumentException if chunkSize is less than zero.
 	 */
-	public Chunk chunk(int chunkSize) throws ReadFailureException;
+	public ChunkingResult chunk(int chunkSize) throws ReadFailureException;
 
 }

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2006-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.execution.step.simple;
+
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.domain.ItemFailureLog;
+
+/**
+ * Default implementation of the {@link ItemFailureLog} interface that
+ * writes all exceptions via commons logging.  Since generics can't be
+ * used to ensure the list contains exceptions, any non exceptions will
+ * be logged out by calling toString on the object.
+ * 
+ * @author Lucas Ward
+ *
+ */
+public class DefaultItemFailureLog implements ItemFailureLog {
+
+	protected static final Log logger = LogFactory
+	.getLog(DefaultItemFailureLog.class);
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.domain.ItemFailureLog#log(java.util.List)
+	 */
+	public void log(List exceptions) {
+		for(Iterator it = exceptions.iterator(); it.hasNext();){
+			Object exception = it.next();
+			try{
+				Throwable t = (Throwable)exception;
+				logger.error("Error encountered during processing", t);
+			}
+			catch(Exception ex){
+				logger.error("Invalid type for logging: [" + exception.toString() + "]");
+			}
+			
+		}
+	}
+
+}

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java

@@ -19,6 +19,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.springframework.batch.core.domain.Chunk;
+import org.springframework.batch.core.domain.ChunkingResult;
 import org.springframework.batch.core.domain.ItemSkipPolicy;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.io.exception.ReadFailureException;
@@ -50,11 +51,12 @@ public class ItemChunker implements Chunker {
 		this.readFailurePolicy = readFailurePolicy;
 	}
 
-	public Chunk chunk(int size) throws ReadFailureException {
+	public ChunkingResult chunk(int size) throws ReadFailureException {
 		Assert.isTrue(size > 0, "Chunk size must be greater than 0");
 
 		int counter = 0;
 		List items = new ArrayList(size);
+		List exceptions = new ArrayList();
 
 		Object item;
 		while (counter < size) {
@@ -66,6 +68,7 @@ public class ItemChunker implements Chunker {
 				items.add(item);
 				counter++;
 			} catch (Exception ex) {
+				exceptions.add(ex);
 				if(!readFailurePolicy.shouldSkip(ex, stepExecution)){
 					rethrow(ex);
 				}
@@ -76,7 +79,7 @@ public class ItemChunker implements Chunker {
 			return null;
 		}
 
-		return new Chunk(getChunkId(), items);
+		return new ChunkingResult(new Chunk(getChunkId(), items), exceptions);
 	}
 	
 	private void rethrow(Exception ex){

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java

@@ -20,10 +20,11 @@ import java.util.Iterator;
 import java.util.List;
 
 import org.springframework.batch.core.domain.Chunk;
-import org.springframework.batch.core.domain.ChunkResult;
+import org.springframework.batch.core.domain.DechunkingResult;
 import org.springframework.batch.core.domain.Dechunker;
 import org.springframework.batch.core.domain.ItemSkipPolicy;
 import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.io.exception.WriteFailureException;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.util.Assert;
 
@@ -48,7 +49,7 @@ public class ItemDechunker implements Dechunker {
 	/* (non-Javadoc)
 	 * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk)
 	 */
-	public ChunkResult dechunk(Chunk chunk) throws Exception {
+	public DechunkingResult dechunk(Chunk chunk) throws Exception {
 		
 		Assert.notNull(chunk, "Chunk must not be null");
 		List skippedItems = new ArrayList();
@@ -61,7 +62,7 @@ public class ItemDechunker implements Dechunker {
 			catch(Exception ex){
 				if(itemSkipPolicy.shouldSkip(ex, stepExecution)){
 					stepExecution.incrementSkipCount();
-					skippedItems.add(item);
+					skippedItems.add(new WriteFailureException(ex, item));
 				}
 				else{
 					rethrow(ex);
@@ -69,7 +70,7 @@ public class ItemDechunker implements Dechunker {
 			}
 		}
 		
-		return new ChunkResult(ChunkResult.SUCCESS, chunk.getId(), skippedItems);
+		return new DechunkingResult(true, chunk.getId(), skippedItems);
 	}
 	
 	public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java

@@ -18,7 +18,6 @@ package org.springframework.batch.execution.step.simple;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.List;
 
 import junit.framework.TestCase;
 
@@ -31,10 +30,6 @@ import org.springframework.batch.core.domain.JobSupport;
 import org.springframework.batch.core.domain.StepExecution;
 import org.springframework.batch.core.domain.StepInstance;
 import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.execution.repository.SimpleJobRepository;
-import org.springframework.batch.execution.repository.dao.MapJobDao;
-import org.springframework.batch.execution.repository.dao.MapStepDao;
-import org.springframework.batch.execution.repository.dao.StepDao;
 import org.springframework.batch.execution.scope.StepScope;
 import org.springframework.batch.execution.scope.StepSynchronizationManager;
 import org.springframework.batch.item.ExecutionAttributes;

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java

@@ -17,7 +17,7 @@ package org.springframework.batch.execution.step.simple;
 
 import junit.framework.TestCase;
 
-import org.springframework.batch.core.domain.Chunk;
+import org.springframework.batch.core.domain.ChunkingResult;
 import org.springframework.batch.core.domain.ItemSkipPolicy;
 import org.springframework.batch.core.domain.StepExecution;
 
@@ -54,15 +54,15 @@ public class ItemChunkerTests extends TestCase {
 	public void testSizePositive() {
 		MockItemReader itemReader = new MockItemReader(10);
 		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		Chunk chunk = chunkReader.chunk(10);
-		assertEquals(10, chunk.getItems().size());
+		ChunkingResult chunkingResult = chunkReader.chunk(10);
+		assertEquals(10, chunkingResult.getChunk().getItems().size());
 	}
 
 	public void testIncompleteChunk() {
 		MockItemReader itemReader = new MockItemReader(5);
 		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		Chunk chunk = chunkReader.chunk(10);
-		assertEquals(5, chunk.getItems().size());
+		ChunkingResult chunkingResult = chunkReader.chunk(10);
+		assertEquals(5, chunkingResult.getChunk().getItems().size());
 	}
 
 	public void testPolicyNoContinue() {
@@ -82,8 +82,8 @@ public class ItemChunkerTests extends TestCase {
 		itemReader.setFail(true);
 		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
 		chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false));
-		Chunk chunk = chunkReader.chunk(1);
-		assertEquals(1, chunk.getItems().size());
+		ChunkingResult chunkingResult = chunkReader.chunk(1);
+		assertEquals(1,chunkingResult.getChunk().getItems().size());
 	}
 
 	private class StubReadFailurePolicy implements ItemSkipPolicy {

spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java

@@ -20,8 +20,9 @@ import java.util.List;
 
 import org.easymock.MockControl;
 import org.springframework.batch.core.domain.Chunk;
-import org.springframework.batch.core.domain.ChunkResult;
+import org.springframework.batch.core.domain.DechunkingResult;
 import org.springframework.batch.core.domain.StepExecution;
+import org.springframework.batch.io.exception.WriteFailureException;
 import org.springframework.batch.item.ItemWriter;
 
 import junit.framework.TestCase;
@@ -71,9 +72,12 @@ public class ItemDechunkerTests extends TestCase {
 		itemWriter.write("2");
 		writerControl.setThrowable(new Exception());
 		writerControl.replay();
-		ChunkResult result = dechunker.dechunk(chunk);
+		DechunkingResult result = dechunker.dechunk(chunk);
 		writerControl.verify();
-		assertEquals("2",result.getSkippedItems().get(0));
+		List exceptions = result.getExceptions();
+		assertEquals(1, exceptions.size());
+		WriteFailureException exception = (WriteFailureException)exceptions.get(0);
+		assertEquals("2",exception.getItem());
 		
 	}