BATCH-220:Moved SimpleJob (formerly simpleTaskletJob) to use the ChunkedStep

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

@@ -21,18 +21,19 @@ 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.Chunker;
 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.SkippedItemHandler;
 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;
 import org.springframework.batch.core.domain.StepInstance;
+import org.springframework.batch.core.domain.StepSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.runtime.ExitStatusExceptionClassifier;
-import org.springframework.batch.core.tasklet.Tasklet;
 import org.springframework.batch.execution.scope.SimpleStepContext;
 import org.springframework.batch.execution.scope.StepContext;
 import org.springframework.batch.execution.scope.StepScope;
@@ -54,6 +55,7 @@ import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.retry.RetryCallback;
 import org.springframework.batch.retry.RetryContext;
 import org.springframework.batch.retry.support.RetryTemplate;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.util.Assert;
 
@@ -61,14 +63,14 @@ import org.springframework.util.Assert;
  * <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}.
+ * while reading (usually skipped unless configured not to) will be logged out via the {@link SkippedItemHandler}.
  * 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>
+ * the {@link SkippedItemHandler}</p>
  * 
  * <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
@@ -95,7 +97,7 @@ import org.springframework.util.Assert;
  * @author Lucas Ward
  * @author Ben Hale
  */
-public class ChunkedStep extends AbstractStep {
+public class ChunkedStep extends StepSupport implements InitializingBean{
 
 	private static final Log logger = LogFactory.getLog(ChunkedStep.class);
 
@@ -109,7 +111,7 @@ public class ChunkedStep extends AbstractStep {
 	// default to checking current thread for interruption.
 	private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy();
 	
-	private ItemFailureLog failureLog = new DefaultItemFailureLog();
+	private SkippedItemHandler failureLog = new DefaultItemFailureLog();
 
 	private StreamManager streamManager;
 
@@ -141,7 +143,7 @@ public class ChunkedStep extends AbstractStep {
 		this.streamManager = streamManager;
 	}
 	
-	public void setFailureLog(ItemFailureLog failureLog) {
+	public void setFailureLog(SkippedItemHandler failureLog) {
 		this.failureLog = failureLog;
 	}
 
@@ -150,10 +152,10 @@ public class ChunkedStep extends AbstractStep {
 	 * 
 	 * @param jobRepository
 	 */
-	public void setRepository(JobRepository jobRepository) {
+	public void setJobRepository(JobRepository jobRepository) {
 		this.jobRepository = jobRepository;
 	}
-
+	
 	/**
 	 * The {@link RepeatOperations} to use for the outer loop of the batch processing. Should be set up by the caller
 	 * through a factory. Defaults to a plain {@link RepeatTemplate}.
@@ -239,7 +241,7 @@ public class ChunkedStep extends AbstractStep {
 			}
 		}
 		
-		
+		Assert.notNull(jobRepository, "JobRepository must not be null");
 	}
 
 	/**
@@ -298,7 +300,7 @@ public class ChunkedStep extends AbstractStep {
 					}
 					
 					final Chunk chunk = chunkingResult.getChunk();
-					failureLog.log(chunkingResult.getExceptions());
+					failureLog.handle(chunkingResult.getExceptions());
 
 					retryTemplate.execute(new RetryCallback(){
 
@@ -364,7 +366,7 @@ public class ChunkedStep extends AbstractStep {
 		try {
 			
 			DechunkingResult chunkResult = dechunker.dechunk(chunk, stepExecution);
-			failureLog.log(chunkResult.getExceptions());
+			failureLog.handle(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

@@ -1,44 +0,0 @@
-/*
- * 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 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;
-
-
-/**
- * Interface defining the contract for reading a chunk. This is most useful when
- * implementing a 'chunk-oriented' approach to processing. Implementors of this
- * class are expected to aggregate the output of an ItemReader into 'chunks'.
- * 
- * @author Ben Hale
- * @author Lucas Ward
- */
-public interface Chunker {
-
-	/**
-	 * 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, StepExecution stepExecution) throws ReadFailureException;
-
-}

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

@@ -20,10 +20,10 @@ import java.util.List;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.core.domain.ItemFailureLog;
+import org.springframework.batch.core.domain.SkippedItemHandler;
 
 /**
- * Default implementation of the {@link ItemFailureLog} interface that
+ * Default implementation of the {@link SkippedItemHandler} 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.
@@ -31,7 +31,7 @@ import org.springframework.batch.core.domain.ItemFailureLog;
  * @author Lucas Ward
  *
  */
-public class DefaultItemFailureLog implements ItemFailureLog {
+public class DefaultItemFailureLog implements SkippedItemHandler {
 
 	protected static final Log logger = LogFactory
 	.getLog(DefaultItemFailureLog.class);
@@ -39,7 +39,7 @@ public class DefaultItemFailureLog implements ItemFailureLog {
 	/* (non-Javadoc)
 	 * @see org.springframework.batch.core.domain.ItemFailureLog#log(java.util.List)
 	 */
-	public void log(List exceptions) {
+	public void handle(List exceptions) {
 		for(Iterator it = exceptions.iterator(); it.hasNext();){
 			Object exception = it.next();
 			try{

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.Chunker;
 import org.springframework.batch.core.domain.ChunkingResult;
 import org.springframework.batch.core.domain.ItemSkipPolicy;
 import org.springframework.batch.core.domain.StepExecution;
@@ -28,7 +29,8 @@ import org.springframework.util.Assert;
 
 /**
  * Implementation of the {@link Chunker} interface that creates chunks from
- * an {@link ItemReader} 
+ * an {@link ItemReader}.  <strong>It does not buffer chunks</strong>.  If 
+ * the underlying reader has been rolled back, and
  * 
  * @author Ben Hale
  * @author Lucas Ward
@@ -93,4 +95,10 @@ public class ItemChunker implements Chunker {
 		return new Long(chunkCounter++);
 	}
 
+	/**
+	 * No-op implementation. 
+	 */
+	public void flush(StepExecution stepExecution) {
+	}
+
 }

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

@@ -85,7 +85,7 @@ public class ChunkedStepTests extends TestCase {
 		step.setItemWriter(processor);
 		step.setItemReader(getReader(strings));
 		step.setJobRepository(new JobRepositorySupport());
-		step.setTransactionManager(transactionManager);
+		step.setStreamManager(new SimpleStreamManager(transactionManager));
 		step.afterPropertiesSet();
 		return step;
 	}
@@ -103,7 +103,7 @@ public class ChunkedStepTests extends TestCase {
 		SimpleStreamManager streamManager = new SimpleStreamManager(transactionManager);
 		streamManager.setUseClassNameAsPrefix(false);
 		chunkedStep.setStreamManager(streamManager);
-		chunkedStep.setRepository(new JobRepositorySupport());
+		chunkedStep.setJobRepository(new JobRepositorySupport());
 		
 		stepInstance = new StepInstance(new Long(9));
 		jobExecutionContext = new JobExecution(jobInstance);
@@ -172,7 +172,7 @@ public class ChunkedStepTests extends TestCase {
 
 		MockControl repoControl = MockControl.createControl(JobRepository.class);
 		JobRepository repository = (JobRepository)repoControl.getMock();
-		chunkedStep.setRepository(repository);
+		chunkedStep.setJobRepository(repository);
 
 //		StepInstance step = new StepInstance(new Long(1));
 //		JobExecution jobExecutionContext = new JobExecution(jobInstance);