BATCH-220:Further refinements to chunk processing. The chunker/dechunker interfaces now accept a StepExecution, so that they don't have to be instantiated anew for each chunk. They can also be wired into the ChunkedStep as well.

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

@@ -25,6 +25,7 @@ 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.ItemSkipPolicy;
 import org.springframework.batch.core.domain.JobInterruptedException;
 import org.springframework.batch.core.domain.StepContribution;
 import org.springframework.batch.core.domain.StepExecution;
@@ -47,6 +48,7 @@ import org.springframework.batch.item.stream.StreamManager;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatCallback;
 import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.batch.repeat.RepeatListener;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.retry.RetryCallback;
@@ -56,16 +58,38 @@ import org.springframework.transaction.TransactionStatus;
 import org.springframework.util.Assert;
 
 /**
- * Simple implementation of executing the step as a set of chunks, each chunk surrounded by a transaction. The structure
- * is therefore that of two nested loops, with transaction boundary around the whole inner loop. The outer loop is
- * controlled by the step operations ({@link #setStepOperations(RepeatOperations)}), and the inner loop by the chunk
- * operations ({@link #setChunkOperations(RepeatOperations)}). The inner loop should always be executed in a single
- * thread, so the chunk operations should not do any concurrent execution. N.B. usually that means that the chunk
- * operations should be a {@link RepeatTemplate} (which is the default).<br/>
+ * <p>Implementation of the {@link Step} interface that deals with input and output as 'chunks'.  Reading is 
+ * delegated to a {@link Chunker} that will read in a {@link Chunk} of items for processing.  The number of
+ * items per chunks is configurable as the chunk size.  Once the chunk has been read, any errors encountered
+ * while reading (usually skipped unless configured not to) will be logged out via the {@link ItemFailureLog}.
+ * The chunk will then be 'dechunked', which in most scenarios will mean delegating to an {@link ItemWriter} 
+ * by writing out one chunk at a time.  The transaction boundary is around this process.  If any errors are 
+ * encountered, the dechunking process will error out, leaving the decision for retrying the chunk up to
+ * a {@link RepeatTemplate}.  This template is configurable, allowing for the number of retries and how long 
+ * to wait between retries (backoff) to be set.  Once dechunking has been finished, any errors not fatal to
+ * the chunk (usually because the error didn't invalidate the transaction) will also be written out via
+ * the {@link ItemFailureLog}</p>
  * 
- * Clients can use interceptors in the step operations to intercept or listen to the iteration on a step-wide basis, for
- * instance to get a callback when the step is complete. Those that want callbacks at the level of an individual tasks,
- * can specify interceptors for the chunk operations.
+ * <p>Clients can use {@link RepeatListener}s in the step operations to intercept or listen to the iteration 
+ * on a step-wide basis, for instance to get a callback when the step is complete.  The open and close methods of
+ * could easily be done with AOP, however, notifications in between complete chunks (before and after) can
+ * be quite useful</p>
+ * 
+ * <p>Repository Usage: The {@link JobRepository} is used extensively to store metadata about the run such as
+ * when the {@link StepExecution} was started, or the commit count.</p>
+ * 
+ * <p>Interruption: At various times while processing, the step will check to see if it has been interrupted 
+ * by calling the {@link StepInterruptionPolicy}.  This policy could check if thread.isInterupted() is true, 
+ * or RepeatContext.isTerminateOnly() is set.  It could even be a check to see if a 'stop file' has been added 
+ * to a particular directory.  If the step should finish, a {@link JobInterruptedException} is thrown, and the
+ * step will clean up, set the status of the {@link StepExecution} to 'STOPPED' and rethrow.</p.
+ * 
+ * <p>ExitStatusClassification: Any number of fatal errors could be thrown during processing.  In general, the
+ * framework must remain fairly dumb as to what error code these exceptions should translate to.  By default
+ * it's a fairly generic 'FATAL_EXECUTION'.  However, this may be insufficient for many scenarios.  If an
+ * enterprise scheduler is used to kick off a batch job, the exit code is the only means of communication as
+ * to what action must be taken.  It may also be the only result that many batch operators see as well.  Therefore,
+ * an {@link ExitStatusExceptionClassifier} may be used to classify an exception to a particular exit code.</p>
  * 
  * @author Dave Syer
  * @author Lucas Ward
@@ -79,6 +103,7 @@ public class ChunkedStep extends AbstractStep {
 
 	private JobRepository jobRepository;
 
+	//default to simple exception classification.
 	private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier();
 
 	// default to checking current thread for interruption.
@@ -89,8 +114,12 @@ public class ChunkedStep extends AbstractStep {
 	private StreamManager streamManager;
 
 	private ItemReader itemReader;
+	private Chunker chunker;
 
 	private ItemWriter itemWriter;
+	private Dechunker dechunker;
+	
+	private ItemSkipPolicy itemSkipPolicy;
 
 	private RetryTemplate retryTemplate = new RetryTemplate();
 	
@@ -168,6 +197,24 @@ public class ChunkedStep extends AbstractStep {
 	public void setItemWriter(ItemWriter itemWriter) {
 		this.itemWriter = itemWriter;
 	}
+	
+	public void setChunker(Chunker chunker) {
+		this.chunker = chunker;
+	}
+	
+	public void setDechunker(Dechunker dechunker) {
+		this.dechunker = dechunker;
+	}
+	
+	/**
+	 * Set the skip policy.  If set, it will be used for both reading
+	 * and writing.
+	 * 
+	 * @param itemSkipPolicy
+	 */
+	public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
+		this.itemSkipPolicy = itemSkipPolicy;
+	}
 
 	/**
 	 * Check mandatory properties (reader and writer).
@@ -175,8 +222,24 @@ public class ChunkedStep extends AbstractStep {
 	 * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
 	 */
 	public void afterPropertiesSet() throws Exception {
-		Assert.notNull(itemReader, "ItemReader must be provided");
-		Assert.notNull(itemWriter, "ItemWriter must be provided");
+		//This is currently a little bit funky, I don't want to require a chunker or
+		//dechunker to be wired in, since the developer should really only be wiring in a
+		//ItemReader and ItemWriter, a namespace should take care of the issue though.
+		if(chunker == null){
+			chunker = new ItemChunker(itemReader);
+			if(itemSkipPolicy != null){
+				((ItemChunker)chunker).setItemSkipPolicy(itemSkipPolicy);
+			}
+		}
+		
+		if(dechunker == null){
+			dechunker = new ItemDechunker(itemWriter);
+			if(itemSkipPolicy != null){
+				((ItemChunker)dechunker).setItemSkipPolicy(itemSkipPolicy);
+			}
+		}
+		
+		
 	}
 
 	/**
@@ -184,7 +247,7 @@ public class ChunkedStep extends AbstractStep {
 	 * into chunks, each one executing in a transaction. The step and its execution and execution context are all given
 	 * an up to date {@link BatchStatus}, and the {@link JobRepository} is used to store the result. Various reporting
 	 * information are also added to the current context (the {@link RepeatContext} governing the step execution, which
-	 * would normally be available to the caller somehow through the step's {@link JobExecutionContext}.<br/>
+	 * would normally be available to the caller somehow through the step's {@link StepContext}.<br/>
 	 * 
 	 * @throws JobInterruptedException if the step or a chunk is interrupted
 	 * @throws RuntimeException if there is an exception during a chunk execution
@@ -228,9 +291,7 @@ public class ChunkedStep extends AbstractStep {
 					// interruption.
 					interruptionPolicy.checkInterrupted(context);
 					
-					//shouldn't have to create a chunker each time, I'll refactor the interface later
-					Chunker chunker = new ItemChunker(itemReader, stepExecution);
-					ChunkingResult chunkingResult = chunker.chunk(chunkSize);
+					ChunkingResult chunkingResult = chunker.chunk(chunkSize, stepExecution);
 					
 					if(chunkingResult == null){
 						return ExitStatus.FINISHED;
@@ -287,11 +348,11 @@ public class ChunkedStep extends AbstractStep {
 	}
 
 	/**
-	 * Execute a bunch of identical business logic operations all within a transaction. The transaction is
-	 * programmatically started and stopped outside this method, so subclasses that override do not need to create a
-	 * transaction.
+	 * Execute a bunch of identical business logic operations all within a transaction.
 	 * 
-	 * @param stepInstance the current step containing the {@link Tasklet} with the business logic.
+	 * @param stepExecution the current execution in which to process the chunk in.
+	 * @param chunk to be processed.
+	 * @param stepContext the current step context.
 	 * @return true if there is more data to process.
 	 */
 	void processChunk(Chunk chunk, final StepExecution stepExecution, StepContext stepContext) {
@@ -301,10 +362,8 @@ public class ChunkedStep extends AbstractStep {
 		final StepContribution contribution = stepExecution.createStepContribution();
 		
 		try {
-
-			Dechunker dechunker = new ItemDechunker(itemWriter, stepExecution);
 			
-			DechunkingResult chunkResult = dechunker.dechunk(chunk);
+			DechunkingResult chunkResult = dechunker.dechunk(chunk, stepExecution);
 			failureLog.log(chunkResult.getExceptions());
 
 			// TODO: check that stepExecution can
@@ -363,7 +422,7 @@ public class ChunkedStep extends AbstractStep {
 		
 	}
 
-	/**
+	/*
 	 * Convenience method to update the status in all relevant places.
 	 * 
 	 * @param stepInstance the current step

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

@@ -17,6 +17,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.core.domain.StepExecution;
 import org.springframework.batch.io.exception.ReadFailureException;
 
 
@@ -31,12 +32,13 @@ import org.springframework.batch.io.exception.ReadFailureException;
 public interface Chunker {
 
 	/**
-	 * Read in a chunk, given the provided chunk size.
+	 * Read in a chunk, given the provided chunk size for the given StepExecution.
 	 * 
 	 * @param chunkSize the number of items that should be read for this chunk.
+	 * @param StepExecution the stepExecution the current chunk is being processed within.
 	 * @return the {@link Chunk} that has been read.
 	 * @throws IllegalArgumentException if chunkSize is less than zero.
 	 */
-	public ChunkingResult chunk(int chunkSize) throws ReadFailureException;
+	public ChunkingResult chunk(int chunkSize, StepExecution stepExecution) throws ReadFailureException;
 
 }

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

@@ -36,22 +36,20 @@ import org.springframework.util.Assert;
 public class ItemChunker implements Chunker {
 
 	private final ItemReader itemReader;
-	private final StepExecution stepExecution;
-
 	private long chunkCounter = 0;
 
-	private ItemSkipPolicy readFailurePolicy = new NeverSkipItemSkipPolicy();
+	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
 
-	public ItemChunker(ItemReader itemReader, StepExecution stepExecution) {
+	public ItemChunker(ItemReader itemReader) {
+		Assert.notNull(itemReader, "ItemReader must not be null");
 		this.itemReader = itemReader;
-		this.stepExecution = stepExecution;
 	}
 
-	public void setReadFailurePolicy(ItemSkipPolicy readFailurePolicy) {
-		this.readFailurePolicy = readFailurePolicy;
+	public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
+		this.itemSkipPolicy = itemSkipPolicy;
 	}
 
-	public ChunkingResult chunk(int size) throws ReadFailureException {
+	public ChunkingResult chunk(int size, StepExecution stepExecution) throws ReadFailureException {
 		Assert.isTrue(size > 0, "Chunk size must be greater than 0");
 
 		int counter = 0;
@@ -69,7 +67,7 @@ public class ItemChunker implements Chunker {
 				counter++;
 			} catch (Exception ex) {
 				exceptions.add(ex);
-				if(!readFailurePolicy.shouldSkip(ex, stepExecution)){
+				if(!itemSkipPolicy.shouldSkip(ex, stepExecution)){
 					rethrow(ex);
 				}
 			}

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

@@ -38,20 +38,19 @@ import org.springframework.util.Assert;
 public class ItemDechunker implements Dechunker {
 
 	private final ItemWriter itemWriter;
-	private final StepExecution stepExecution;
 	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
 		
-	public ItemDechunker(ItemWriter itemWriter, StepExecution stepExecution) {
+	public ItemDechunker(ItemWriter itemWriter) {
 		this.itemWriter = itemWriter;
-		this.stepExecution = stepExecution;
 	}
 	
 	/* (non-Javadoc)
 	 * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk)
 	 */
-	public DechunkingResult dechunk(Chunk chunk) throws Exception {
+	public DechunkingResult dechunk(Chunk chunk, StepExecution stepExecution) throws Exception {
 		
 		Assert.notNull(chunk, "Chunk must not be null");
+		Assert.notNull(stepExecution, "StepExecution must not be null");
 		List skippedItems = new ArrayList();
 		for(Iterator it = chunk.getItems().iterator(); it.hasNext();){
 			

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

@@ -123,7 +123,7 @@ public class ChunkedStepTests extends TestCase {
 		final JobExecution jobExecution = new JobExecution(jobInstance);
 		final StepExecution stepExecution = new StepExecution(step, jobExecution);
 
-		chunkedStep.setItemReader(new ItemReader() {
+		chunkedStep.setChunker(new ItemChunker(new ItemReader() {
 			int counter = 0;
 			public Object read() throws Exception {
 				assertEquals(step, stepExecution.getStep());
@@ -135,7 +135,7 @@ public class ChunkedStepTests extends TestCase {
 					return null;
 				}
 			}
-		});
+		}));
 
 		chunkedStep.execute(stepExecution);
 		assertEquals(2, processed.size());
@@ -212,7 +212,7 @@ public class ChunkedStepTests extends TestCase {
 
 		};
 
-		chunkedStep.setItemReader(itemReader);
+		chunkedStep.setChunker(new ItemChunker(itemReader));
 
 		try {
 			chunkedStep.execute(stepExecution);
@@ -233,7 +233,7 @@ public class ChunkedStepTests extends TestCase {
 			}
 		};
 		
-		chunkedStep.setItemWriter(itemWriter);
+		chunkedStep.setDechunker(new ItemDechunker(itemWriter));
 		
 		try{
 			chunkedStep.execute(stepExecution);

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

@@ -34,8 +34,8 @@ public class ItemChunkerTests extends TestCase {
 	public void testSizeNegative() {
 		try {
 			MockItemReader itemReader = new MockItemReader(10);
-			ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-			chunkReader.chunk(-1);
+			ItemChunker chunkReader = new ItemChunker(itemReader);
+			chunkReader.chunk(-1, stepExecution);
 			fail();
 		} catch (IllegalArgumentException e) {
 		}
@@ -44,8 +44,8 @@ public class ItemChunkerTests extends TestCase {
 	public void testSizeZero() {
 		try {
 			MockItemReader itemReader = new MockItemReader(10);
-			ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-			chunkReader.chunk(0);
+			ItemChunker chunkReader = new ItemChunker(itemReader);
+			chunkReader.chunk(0, stepExecution);
 			fail();
 		} catch (IllegalArgumentException e) {
 		}
@@ -53,25 +53,25 @@ public class ItemChunkerTests extends TestCase {
 
 	public void testSizePositive() {
 		MockItemReader itemReader = new MockItemReader(10);
-		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		ChunkingResult chunkingResult = chunkReader.chunk(10);
+		ItemChunker chunkReader = new ItemChunker(itemReader);
+		ChunkingResult chunkingResult = chunkReader.chunk(10, stepExecution);
 		assertEquals(10, chunkingResult.getChunk().getItems().size());
 	}
 
 	public void testIncompleteChunk() {
 		MockItemReader itemReader = new MockItemReader(5);
-		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		ChunkingResult chunkingResult = chunkReader.chunk(10);
+		ItemChunker chunkReader = new ItemChunker(itemReader);
+		ChunkingResult chunkingResult = chunkReader.chunk(10, stepExecution);
 		assertEquals(5, chunkingResult.getChunk().getItems().size());
 	}
 
 	public void testPolicyNoContinue() {
 		MockItemReader itemReader = new MockItemReader(1);
 		itemReader.setFail(true);
-		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(true));
+		ItemChunker chunkReader = new ItemChunker(itemReader);
+		chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(true));
 		try {
-			chunkReader.chunk(10);
+			chunkReader.chunk(10, stepExecution);
 			fail();
 		} catch (RuntimeException e) {
 		}
@@ -80,9 +80,9 @@ public class ItemChunkerTests extends TestCase {
 	public void testPolicyContinueWithFailure() {
 		MockItemReader itemReader = new MockItemReader(1);
 		itemReader.setFail(true);
-		ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution);
-		chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false));
-		ChunkingResult chunkingResult = chunkReader.chunk(1);
+		ItemChunker chunkReader = new ItemChunker(itemReader);
+		chunkReader.setItemSkipPolicy(new StubReadFailurePolicy(false));
+		ChunkingResult chunkingResult = chunkReader.chunk(1, stepExecution);
 		assertEquals(1,chunkingResult.getChunk().getItems().size());
 	}
 

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

@@ -48,7 +48,7 @@ public class ItemDechunkerTests extends TestCase {
 		
 		itemWriter = (ItemWriter)writerControl.getMock();
 		stepExecution = new StepExecution(null,null);
-		dechunker = new ItemDechunker(itemWriter, stepExecution);
+		dechunker = new ItemDechunker(itemWriter);
 		List items = new ArrayList();
 		items.add("1");
 		items.add("2");
@@ -61,7 +61,7 @@ public class ItemDechunkerTests extends TestCase {
 		itemWriter.write("1");
 		itemWriter.write("2");
 		writerControl.replay();
-		dechunker.dechunk(chunk);
+		dechunker.dechunk(chunk, stepExecution);
 		writerControl.verify();
 	}
 	
@@ -72,7 +72,7 @@ public class ItemDechunkerTests extends TestCase {
 		itemWriter.write("2");
 		writerControl.setThrowable(new Exception());
 		writerControl.replay();
-		DechunkingResult result = dechunker.dechunk(chunk);
+		DechunkingResult result = dechunker.dechunk(chunk, stepExecution);
 		writerControl.verify();
 		List exceptions = result.getExceptions();
 		assertEquals(1, exceptions.size());
@@ -87,7 +87,7 @@ public class ItemDechunkerTests extends TestCase {
 		writerControl.setThrowable(new NullPointerException());
 		writerControl.replay();
 		try{
-			dechunker.dechunk(chunk);
+			dechunker.dechunk(chunk, stepExecution);
 			fail();
 		}
 		catch(NullPointerException ex){