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}
+ * the {@link SkippedItemHandler} * *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 diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java deleted file mode 100644 index 71e4bf4f2..000000000 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/Chunker.java +++ /dev/null @@ -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; - -} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java index ed1a9f240..85139cfd3 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/DefaultItemFailureLog.java +++ b/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{ diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java index 6a60251d0..60c4293a1 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemChunker.java +++ b/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}. It does not buffer chunks. 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) { + } + } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java index 5b5eac985..c797a1f2d 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ChunkedStepTests.java +++ b/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);