diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipItemSkipPolicy.java similarity index 71% rename from spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java rename to spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipItemSkipPolicy.java index de35c6ea5..3edd39167 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipReadFailurePolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AlwaysSkipItemSkipPolicy.java @@ -15,19 +15,19 @@ */ package org.springframework.batch.execution.step.simple; -import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; /** - * Implementation of the {@link ReadFailurePolicy} interface that - * will always return that reading should continue. + * Implementation of the {@link ItemSkipPolicy} interface that + * will always return that an item should be skipped. * * @author Ben Hale * @author Lucas Ward */ -public class AlwaysSkipReadFailurePolicy implements ReadFailurePolicy { +public class AlwaysSkipItemSkipPolicy implements ItemSkipPolicy { - public boolean shouldContinue(Exception ex, StepExecution stepExecution) { + public boolean shouldSkip(Exception ex, StepExecution stepExecution) { return true; } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStepExecutor.java new file mode 100644 index 000000000..1d2a9c9e8 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStepExecutor.java @@ -0,0 +1,450 @@ +/* + * Copyright 2006-2007 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.Date; + +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.Dechunker; +import org.springframework.batch.core.domain.JobInterruptedException; +import org.springframework.batch.core.domain.Step; +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.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; +import org.springframework.batch.execution.scope.StepSynchronizationManager; +import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.ExecutionAttributes; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.KeyedItemReader; +import org.springframework.batch.item.exception.ResetFailedException; +import org.springframework.batch.item.stream.SimpleStreamManager; +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.RepeatOperations; +import org.springframework.batch.repeat.exception.handler.ExceptionHandler; +import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.callback.ItemReaderRetryCallback; +import org.springframework.batch.retry.policy.ItemReaderRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.beans.factory.InitializingBean; +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).
+ * + * 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. + * + * @author Dave Syer + * @author Lucas Ward + * @author Ben Hale + */ +public class ChunkedStepExecutor implements InitializingBean { + + private static final Log logger = LogFactory.getLog(ChunkedStepExecutor.class); + + private RepeatOperations chunkOperations = new RepeatTemplate(); + + private RepeatOperations stepOperations = new RepeatTemplate(); + + private JobRepository jobRepository; + + private ExitStatusExceptionClassifier exceptionClassifier = new SimpleExitStatusExceptionClassifier(); + + // default to checking current thread for interruption. + private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); + + private AbstractStep step; + + private StreamManager streamManager; + + private ItemReader itemReader; + + private ItemWriter itemWriter; + + private RetryPolicy retryPolicy = null; + + private ItemRecoverer itemRecoverer; + + private RetryTemplate template = new RetryTemplate(); + + private ItemReaderRetryCallback retryCallback; + + /** + * Package private constructor so the step can create a the executor. + */ + ChunkedStepExecutor(AbstractStep abstractStep) { + this.step = abstractStep; + } + + /** + * Public setter for the {@link StreamManager}. This will be used to create the {@link StepContext}, and hence any + * component that is a {@link ItemStream} and in step scope will be registered with the service. The + * {@link StepContext} is then a source of aggregate statistics for the step. + * + * @param streamManager the {@link StreamManager} to set. Default is a {@link SimpleStreamManager}. + */ + public void setStreamManager(StreamManager streamManager) { + this.streamManager = streamManager; + } + + /** + * Injected strategy for storage and retrieval of persistent step information. Mandatory property. + * + * @param jobRepository + */ + public void setRepository(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}. + * + * @param stepOperations a {@link RepeatOperations} instance. + */ + public void setStepOperations(RepeatOperations stepOperations) { + this.stepOperations = stepOperations; + } + + /** + * The {@link RepeatOperations} to use for the inner loop of the batch processing. Should be set up by the caller + * through a factory. Defaults to a plain {@link RepeatTemplate}. + * + * @param chunkOperations a {@link RepeatOperations} instance. + */ + public void setChunkOperations(RepeatOperations chunkOperations) { + this.chunkOperations = chunkOperations; + } + + /** + * Setter for the {@link StepInterruptionPolicy}. The policy is used to check whether an external request has been + * made to interrupt the job execution. + * + * @param interruptionPolicy a {@link StepInterruptionPolicy} + */ + public void setInterruptionPolicy(StepInterruptionPolicy interruptionPolicy) { + this.interruptionPolicy = interruptionPolicy; + } + + /** + * Setter for the {@link ExitStatusExceptionClassifier} that will be used to classify any exception that causes a job + * to fail. + * + * @param exceptionClassifier + */ + public void setExceptionClassifier(ExitStatusExceptionClassifier exceptionClassifier) { + this.exceptionClassifier = exceptionClassifier; + } + + /** + * @param itemReader + */ + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * @param itemWriter + */ + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + /** + * Setter for injecting optional recovery handler. + * + * @param itemRecoverer + */ + public void setItemRecoverer(ItemRecoverer itemRecoverer) { + this.itemRecoverer = itemRecoverer; + } + + /** + * Public setter for the retryPolicy. + * + * @param retyPolicy the retryPolicy to set + */ + public void setRetryPolicy(RetryPolicy retryPolicy) { + this.retryPolicy = retryPolicy; + } + + /** + * Check mandatory properties (reader and writer). + * + * @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"); + + if (itemRecoverer == null && (itemReader instanceof ItemRecoverer)) { + itemRecoverer = (ItemRecoverer) itemReader; + } + + ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy); + template.setRetryPolicy(itemProviderRetryPolicy); + + if (retryPolicy != null) { + Assert.state(itemReader instanceof KeyedItemReader, + "ItemReader must be instance of KeyedItemReader to use the retry policy"); + retryCallback = new ItemReaderRetryCallback((KeyedItemReader) itemReader, itemWriter); + retryCallback.setRecoverer(itemRecoverer); + } + + } + + /** + * Apply the configuration by inspecting it to see if it has any relevant policy information. + * + * @param step a step + */ + void applyConfiguration(AbstractStep step) { + + if (step instanceof SimpleStep) { + SimpleStep simple = (SimpleStep) step; + if (this.chunkOperations instanceof RepeatTemplate) { + RepeatTemplate template = (RepeatTemplate) this.chunkOperations; + template.setCompletionPolicy(new SimpleCompletionPolicy(simple.getCommitInterval())); + } + } + + ExceptionHandler exceptionHandler = step.getExceptionHandler(); + + if (step.getSkipLimit() > 0 && exceptionHandler == null) { + SimpleLimitExceptionHandler handler = new SimpleLimitExceptionHandler(); + handler.setLimit(step.getSkipLimit()); + exceptionHandler = handler; + } + + if (this.stepOperations instanceof RepeatTemplate && exceptionHandler != null) { + RepeatTemplate template = (RepeatTemplate) this.stepOperations; + template.setExceptionHandler(exceptionHandler); + } + + } + + /** + * Process the step and update its context so that progress can be monitored by the caller. The step is broken down + * 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}.
+ * + * @throws JobInterruptedException if the step or a chunk is interrupted + * @throws RuntimeException if there is an exception during a chunk execution + * @see StepExecutor#execute(StepExecution) + */ + public void execute(final StepExecution stepExecution) throws BatchCriticalException, JobInterruptedException { + + final StepInstance stepInstance = stepExecution.getStep(); + Assert.notNull(stepInstance); + boolean isRestart = stepInstance.getStepExecutionCount() > 0 ? true : false; + + ExitStatus status = ExitStatus.FAILED; + + StepContext parentStepContext = StepSynchronizationManager.getContext(); + final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager); + StepSynchronizationManager.register(stepContext); + // Add the job identifier so that it can be used to identify + // the conversation in StepScope + stepContext.setAttribute(StepScope.ID_KEY, stepExecution.getJobExecution().getId()); + + final boolean saveExecutionAttributes = step.isSaveExecutionAttributes(); + + if (saveExecutionAttributes && isRestart && stepInstance.getLastExecution() != null) { + stepExecution.setExecutionAttributes(stepInstance.getLastExecution().getExecutionAttributes()); + stepContext.restoreFrom(stepExecution.getExecutionAttributes()); + } + + try { + + stepExecution.setStartTime(new Date(System.currentTimeMillis())); + stepInstance.setLastExecution(stepExecution); + updateStatus(stepExecution, BatchStatus.STARTED); + + status = stepOperations.iterate(new RepeatCallback() { + + public ExitStatus doInIteration(final RepeatContext context) throws Exception { + + + // Before starting a new transaction, check for + // interruption. + interruptionPolicy.checkInterrupted(context); + + ExitStatus result = processChunk(step, stepExecution, stepContext); + + + // Check for interruption after transaction as well, so that + // the interrupted exception is correctly propagated up to + // caller + interruptionPolicy.checkInterrupted(context); + + return result; + + } + }); + + updateStatus(stepExecution, BatchStatus.COMPLETED); + } catch (RuntimeException e) { + + // classify exception so an exit code can be stored. + status = exceptionClassifier.classifyForExitCode(e); + if (e.getCause() instanceof JobInterruptedException) { + updateStatus(stepExecution, BatchStatus.STOPPED); + throw (JobInterruptedException) e.getCause(); + } else if (e instanceof ResetFailedException) { + updateStatus(stepExecution, BatchStatus.UNKNOWN); + throw (ResetFailedException) e; + } else { + updateStatus(stepExecution, BatchStatus.FAILED); + throw e; + } + + } finally { + stepExecution.setExitStatus(status); + stepExecution.setEndTime(new Date(System.currentTimeMillis())); + try { + jobRepository.saveOrUpdate(stepExecution); + } finally { + // clear any registered synchronizations + StepSynchronizationManager.close(); + } + } + + } + + /** + * 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. + * + * @param step the current step containing the {@link Tasklet} with the business logic. + * @return true if there is more data to process. + */ + ExitStatus processChunk(final Step step, final StepExecution stepExecution, StepContext stepContext) { + + TransactionStatus transaction = streamManager.getTransaction(stepContext); + + final StepContribution contribution = stepExecution.createStepContribution(); + + try { + + //result = processChunk(step, contribution); + + Chunker chunker = new ItemChunker(itemReader, stepExecution); + Dechunker dechunker = new ItemDechunker(itemWriter, stepExecution); + //should be able to get the chunk size from the step, temporarily hardcoding at 5 + Chunk chunk = chunker.chunk(5); + ChunkResult chunkResult = dechunker.dechunk(chunk); + + // TODO: check that stepExecution can + // aggregate these contributions if they + // come in asynchronously. + ExecutionAttributes statistics = stepContext.getExecutionAttributes(); + contribution.setExecutionAttributes(statistics); + contribution.incrementCommitCount(); + + // If the step operations are asynchronous then we need + // to synchronize changes to the step execution (at a + // minimum). + synchronized (stepExecution) { + + // Apply the contribution to the step + // only if chunk was successful + stepExecution.apply(contribution); + + if (step.isSaveExecutionAttributes()) { + stepExecution.setExecutionAttributes(stepContext.getExecutionAttributes()); + } + jobRepository.saveOrUpdate(stepExecution); + + } + + streamManager.commit(transaction); + + } catch (Throwable t) { + /* + * Any exception thrown within the transaction template will automatically cause the transaction + * to rollback. We need to include exceptions during an attempted commit (e.g. Hibernate flush) + * so this catch block comes outside the transaction. + */ + synchronized (stepExecution) { + stepExecution.rollback(); + } + try { + streamManager.rollback(transaction); + } catch (ResetFailedException e) { + // The original Throwable cause is in danger of + // being lost here, so we log the reset + // failure and re-throw with cause of the rollback. + logger.error("Encountered reset error on rollback: " + + "one of the streams may be in an inconsistent state, " + + "so this step should not proceed", e); + throw new ResetFailedException("Encountered reset error on rollback. " + + "Consult logs for the cause of the reet failure. " + + "The cause of the original rollback is incuded here.", t); + } + if (t instanceof RuntimeException) { + throw (RuntimeException) t; + } else { + throw new RuntimeException(t); + } + } + + return null; + + } + + /** + * Convenience method to update the status in all relevant places. + * + * @param step the current step + * @param stepExecution the current stepExecution + * @param status the status to set + */ + private void updateStatus(StepExecution stepExecution, BatchStatus status) { + StepInstance step = stepExecution.getStep(); + stepExecution.setStatus(status); + jobRepository.update(step); + jobRepository.saveOrUpdate(stepExecution); + } +} 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 index 45fc52acc..b2c3aed86 100644 --- 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 @@ -36,6 +36,6 @@ public interface Chunker { * @return the {@link Chunk} that has been read. * @throws IllegalArgumentException if chunkSize is less than zero. */ - public Chunk read(int chunkSize) throws ReadFailureException; + public Chunk chunk(int chunkSize) throws ReadFailureException; } 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 374a4c8c9..7ad5646d9 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,7 +19,7 @@ import java.util.ArrayList; import java.util.List; import org.springframework.batch.core.domain.Chunk; -import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.ReadFailureException; import org.springframework.batch.item.ItemReader; @@ -39,18 +39,18 @@ public class ItemChunker implements Chunker { private long chunkCounter = 0; - private ReadFailurePolicy readFailurePolicy = new AlwaysSkipReadFailurePolicy(); + private ItemSkipPolicy readFailurePolicy = new AlwaysSkipItemSkipPolicy(); public ItemChunker(ItemReader itemReader, StepExecution stepExecution) { this.itemReader = itemReader; this.stepExecution = stepExecution; } - public void setReadFailurePolicy(ReadFailurePolicy readFailurePolicy) { + public void setReadFailurePolicy(ItemSkipPolicy readFailurePolicy) { this.readFailurePolicy = readFailurePolicy; } - public Chunk read(int size) throws ReadFailureException { + public Chunk chunk(int size) throws ReadFailureException { Assert.isTrue(size > 0, "Chunk size must be greater than 0"); int counter = 0; @@ -66,7 +66,7 @@ public class ItemChunker implements Chunker { items.add(item); counter++; } catch (Exception ex) { - if(!readFailurePolicy.shouldContinue(ex, stepExecution)){ + if(!readFailurePolicy.shouldSkip(ex, stepExecution)){ rethrow(ex); } } diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java new file mode 100644 index 000000000..c8edbd060 --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java @@ -0,0 +1,88 @@ +/* + * 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.ArrayList; +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.Dechunker; +import org.springframework.batch.core.domain.ItemSkipPolicy; +import org.springframework.batch.core.domain.StepExecution; +import org.springframework.batch.item.ItemWriter; +import org.springframework.util.Assert; + +/** + * Implementation of the {@link Dechunker} interface that passes items to an + * {@link ItemWriter} one at a time. + * + * @author Lucas Ward + * + */ +public class ItemDechunker implements Dechunker { + + private final ItemWriter itemWriter; + private final StepExecution stepExecution; + private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); + + public ItemDechunker(ItemWriter itemWriter, StepExecution stepExecution) { + this.itemWriter = itemWriter; + this.stepExecution = stepExecution; + } + + /* (non-Javadoc) + * @see org.springframework.batch.core.domain.Dechunker#dechunk(org.springframework.batch.core.domain.Chunk) + */ + public ChunkResult dechunk(Chunk chunk) throws Exception { + + Assert.notNull(chunk, "Chunk must not be null"); + List skippedItems = new ArrayList(); + for(Iterator it = chunk.getItems().iterator(); it.hasNext();){ + + Object item = it.next(); + try{ + itemWriter.write(item); + } + catch(Exception ex){ + if(itemSkipPolicy.shouldSkip(ex, stepExecution)){ + stepExecution.incrementSkipCount(); + skippedItems.add(item); + } + else{ + rethrow(ex); + } + } + } + + return new ChunkResult(ChunkResult.SUCCESS, chunk.getId(), skippedItems); + } + + public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) { + this.itemSkipPolicy = itemSkipPolicy; + } + + private void rethrow(Exception ex){ + if(ex instanceof RuntimeException){ + throw (RuntimeException)ex; + } + else{ + throw new RuntimeException("Error encountered while dechunking", ex); + } + } + +} diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/LimitCheckingItemSkipPolicy.java similarity index 86% rename from spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java rename to spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/LimitCheckingItemSkipPolicy.java index d6f1a4197..fe8a61d80 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicy.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/LimitCheckingItemSkipPolicy.java @@ -19,12 +19,12 @@ import java.io.FileNotFoundException; import java.util.ArrayList; import java.util.List; -import org.springframework.batch.core.domain.ReadFailurePolicy; +import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.io.exception.FlatFileParsingException; /** - *

{@link ReadFailurePolicy} that determines whether or not reading should + *

{@link ItemSkipPolicy} that determines whether or not reading should * continue based upon how many items have been skipped. This is extremely * useful behavior, as it allows you to skip records, but will throw a * {@link SkipLimitExceededException} if a set limit has been exceeded. For example, @@ -41,17 +41,17 @@ import org.springframework.batch.io.exception.FlatFileParsingException; * @author Ben Hale * @author Lucas Ward */ -public class SkipLimitReadFailurePolicy implements ReadFailurePolicy { +public class LimitCheckingItemSkipPolicy implements ItemSkipPolicy { private final int skipLimit; private final List skippableExceptions; - public SkipLimitReadFailurePolicy(int skipLimit) { + public LimitCheckingItemSkipPolicy(int skipLimit) { this(skipLimit, new ArrayList(0)); } - public SkipLimitReadFailurePolicy(int skipLimit, List skippableExceptions) { + public LimitCheckingItemSkipPolicy(int skipLimit, List skippableExceptions) { this.skipLimit = skipLimit; this.skippableExceptions = skippableExceptions; } @@ -64,7 +64,7 @@ public class SkipLimitReadFailurePolicy implements ReadFailurePolicy { * is greater than the skipLimit, then a {@link SkipLimitExceededException} * will be thrown. */ - public boolean shouldContinue(Exception ex, StepExecution stepExecution){ + public boolean shouldSkip(Exception ex, StepExecution stepExecution){ if(skippableExceptions.contains(ex.getClass())){ if(stepExecution.getSkipCount() < skipLimit){ stepExecution.incrementSkipCount(); diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/NeverSkipItemSkipPolicy.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/NeverSkipItemSkipPolicy.java new file mode 100644 index 000000000..5ac6ba19b --- /dev/null +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/NeverSkipItemSkipPolicy.java @@ -0,0 +1,34 @@ +/* + * 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.ItemSkipPolicy; +import org.springframework.batch.core.domain.StepExecution; + +/** + * {@link ItemSkipPolicy} implementation that always returns false, + * indicating that an item should not be skipped. + * + * @author Lucas Ward + */ +public class NeverSkipItemSkipPolicy implements ItemSkipPolicy{ + + public boolean shouldSkip(Exception ex, StepExecution stepExecution) { + return false; + } + + +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java index 2fabab361..dadd3b7c7 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemChunkerTests.java @@ -18,7 +18,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.ReadFailurePolicy; +import org.springframework.batch.core.domain.ItemSkipPolicy; import org.springframework.batch.core.domain.StepExecution; public class ItemChunkerTests extends TestCase { @@ -35,7 +35,7 @@ public class ItemChunkerTests extends TestCase { try { MockItemReader itemReader = new MockItemReader(10); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.read(-1); + chunkReader.chunk(-1); fail(); } catch (IllegalArgumentException e) { } @@ -45,7 +45,7 @@ public class ItemChunkerTests extends TestCase { try { MockItemReader itemReader = new MockItemReader(10); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - chunkReader.read(0); + chunkReader.chunk(0); fail(); } catch (IllegalArgumentException e) { } @@ -54,14 +54,14 @@ public class ItemChunkerTests extends TestCase { public void testSizePositive() { MockItemReader itemReader = new MockItemReader(10); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - Chunk chunk = chunkReader.read(10); + Chunk chunk = chunkReader.chunk(10); assertEquals(10, chunk.getItems().size()); } public void testIncompleteChunk() { MockItemReader itemReader = new MockItemReader(5); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); - Chunk chunk = chunkReader.read(10); + Chunk chunk = chunkReader.chunk(10); assertEquals(5, chunk.getItems().size()); } @@ -71,7 +71,7 @@ public class ItemChunkerTests extends TestCase { ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(true)); try { - chunkReader.read(10); + chunkReader.chunk(10); fail(); } catch (RuntimeException e) { } @@ -82,11 +82,11 @@ public class ItemChunkerTests extends TestCase { itemReader.setFail(true); ItemChunker chunkReader = new ItemChunker(itemReader,stepExecution); chunkReader.setReadFailurePolicy(new StubReadFailurePolicy(false)); - Chunk chunk = chunkReader.read(1); + Chunk chunk = chunkReader.chunk(1); assertEquals(1, chunk.getItems().size()); } - private class StubReadFailurePolicy implements ReadFailurePolicy { + private class StubReadFailurePolicy implements ItemSkipPolicy { private final boolean fail; @@ -94,7 +94,7 @@ public class ItemChunkerTests extends TestCase { this.fail = fail; } - public boolean shouldContinue(Exception ex, StepExecution stepExecution) { + public boolean shouldSkip(Exception ex, StepExecution stepExecution) { return !fail; } } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java new file mode 100644 index 000000000..407316660 --- /dev/null +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java @@ -0,0 +1,93 @@ +/* + * 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.ArrayList; +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.StepExecution; +import org.springframework.batch.item.ItemWriter; + +import junit.framework.TestCase; + +/** + * @author Lucas Ward + * + */ +public class ItemDechunkerTests extends TestCase { + + private ItemDechunker dechunker; + private StepExecution stepExecution; + private Chunk chunk; + private ItemWriter itemWriter; + private MockControl writerControl = MockControl.createControl(ItemWriter.class); + + + /* (non-Javadoc) + * @see junit.framework.TestCase#setUp() + */ + protected void setUp() throws Exception { + super.setUp(); + + itemWriter = (ItemWriter)writerControl.getMock(); + stepExecution = new StepExecution(null,null); + dechunker = new ItemDechunker(itemWriter, stepExecution); + List items = new ArrayList(); + items.add("1"); + items.add("2"); + chunk = new Chunk(new Long(1),items); + } + + + public void testNormalProcessing() throws Exception{ + + itemWriter.write("1"); + itemWriter.write("2"); + writerControl.replay(); + dechunker.dechunk(chunk); + writerControl.verify(); + } + + public void testSkip() throws Exception{ + + dechunker.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + itemWriter.write("1"); + itemWriter.write("2"); + writerControl.setThrowable(new Exception()); + writerControl.replay(); + ChunkResult result = dechunker.dechunk(chunk); + writerControl.verify(); + assertEquals("2",result.getSkippedItems().get(0)); + + } + + public void testFailure() throws Exception{ + itemWriter.write("1"); + itemWriter.write("2"); + writerControl.setThrowable(new NullPointerException()); + writerControl.replay(); + try{ + dechunker.dechunk(chunk); + fail(); + } + catch(NullPointerException ex){ + //expected + } + } +} diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java index 58482439c..8a0b02674 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SkipLimitReadFailurePolicyTests.java @@ -30,7 +30,7 @@ import junit.framework.TestCase; */ public class SkipLimitReadFailurePolicyTests extends TestCase { - SkipLimitReadFailurePolicy failurePolicy; + LimitCheckingItemSkipPolicy failurePolicy; StepExecution stepExecution; protected void setUp() throws Exception { @@ -39,14 +39,14 @@ public class SkipLimitReadFailurePolicyTests extends TestCase { List skippableExceptions = new ArrayList(); skippableExceptions.add(FlatFileParsingException.class); - failurePolicy = new SkipLimitReadFailurePolicy(1, skippableExceptions); + failurePolicy = new LimitCheckingItemSkipPolicy(1, skippableExceptions); stepExecution = new StepExecution(null, null); stepExecution.setSkipCount(2); } public void testLimitExceed(){ try{ - failurePolicy.shouldContinue(new FlatFileParsingException("", ""), stepExecution); + failurePolicy.shouldSkip(new FlatFileParsingException("", ""), stepExecution); fail(); } catch(SkipLimitExceededException ex){ @@ -55,12 +55,12 @@ public class SkipLimitReadFailurePolicyTests extends TestCase { } public void testNonSkippableException(){ - assertFalse(failurePolicy.shouldContinue(new FileNotFoundException(), stepExecution)); + assertFalse(failurePolicy.shouldSkip(new FileNotFoundException(), stepExecution)); } public void testSkip(){ stepExecution.setSkipCount(0); - assertTrue(failurePolicy.shouldContinue(new FlatFileParsingException("",""), stepExecution)); + assertTrue(failurePolicy.shouldSkip(new FlatFileParsingException("",""), stepExecution)); } }