From 10e75142905b52f72b0bfda3db823e536c58038e Mon Sep 17 00:00:00 2001 From: lucasward Date: Tue, 12 Feb 2008 06:07:33 +0000 Subject: [PATCH] BATCH-220:Added the 'Dechunker', as the opposite of Chunker. It assumes that some higher level class will be managing transactions. Also renamed ReadFailurePolicy to ItemSkipPolicy (Same name used by Ben in the Sandbox) once I realized that a 'ReadFailurePolicy' and 'WriteFailurePolicy' would have pretty much the same interface. --- ...icy.java => AlwaysSkipItemSkipPolicy.java} | 10 +- .../step/simple/ChunkedStepExecutor.java | 450 ++++++++++++++++++ .../batch/execution/step/simple/Chunker.java | 2 +- .../execution/step/simple/ItemChunker.java | 10 +- .../execution/step/simple/ItemDechunker.java | 88 ++++ ....java => LimitCheckingItemSkipPolicy.java} | 12 +- .../step/simple/NeverSkipItemSkipPolicy.java | 34 ++ .../step/simple/ItemChunkerTests.java | 18 +- .../step/simple/ItemDechunkerTests.java | 93 ++++ .../SkipLimitReadFailurePolicyTests.java | 10 +- 10 files changed, 696 insertions(+), 31 deletions(-) rename spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/{AlwaysSkipReadFailurePolicy.java => AlwaysSkipItemSkipPolicy.java} (71%) create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ChunkedStepExecutor.java create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/ItemDechunker.java rename spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/{SkipLimitReadFailurePolicy.java => LimitCheckingItemSkipPolicy.java} (86%) create mode 100644 spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/NeverSkipItemSkipPolicy.java create mode 100644 spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/ItemDechunkerTests.java 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)); } }