diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java deleted file mode 100644 index ec210bea6..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java +++ /dev/null @@ -1,287 +0,0 @@ -package org.springframework.batch.core.step.item; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; - -import org.springframework.batch.core.SkipListener; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.skip.SkipPolicy; -import org.springframework.batch.core.step.skip.SkipListenerFailedException; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.retry.RecoveryCallback; -import org.springframework.batch.retry.RetryCallback; -import org.springframework.batch.retry.RetryContext; -import org.springframework.batch.retry.RetryException; -import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.retry.support.DefaultRetryState; -import org.springframework.batch.support.Classifier; -import org.springframework.core.AttributeAccessor; - -/** - * Fault-tolerant implementation of the process and write phase of chunk - * processing. - * - * @param input item type - * @param output item type - * - * @see FaultTolerantChunkOrientedTasklet - * @see NonbufferingFaultTolerantChunkOrientedTasklet - * - * @author Robert Kasanicky - */ -public abstract class AbstractFaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { - - final static protected String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_KEY"; - - final static protected String SKIPPED_OUTPUTS_KEY = "SKIPPED_OUTPUTS_KEY"; - - final static protected String SKIPPED_READS_KEY = "SKIPPED_READS_KEY"; - - final private RetryOperations retryOperations; - - final private RepeatOperations repeatOperations; - - final private SkipPolicy writeSkipPolicy; - - final private SkipPolicy processSkipPolicy; - - final private SkipPolicy readSkipPolicy; - - final private Classifier rollbackClassifier; - - public AbstractFaultTolerantChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, - RetryOperations retryOperations, SkipPolicy readSkipPolicy, SkipPolicy processSkipPolicy, - SkipPolicy writeSkipPolicy, Classifier rollbackClassifier, - RepeatOperations repeatTemplate) { - - super(itemReader, itemProcessor, itemWriter); - this.retryOperations = retryOperations; - this.readSkipPolicy = readSkipPolicy; - this.processSkipPolicy = processSkipPolicy; - this.writeSkipPolicy = writeSkipPolicy; - this.rollbackClassifier = rollbackClassifier; - this.repeatOperations = repeatTemplate; - } - - protected SkipPolicy getReadSkipPolicy() { - return readSkipPolicy; - } - - protected RepeatOperations getRepeatOperations() { - return repeatOperations; - } - - /** - * Call all skip listeners in read-process-write order - * @param skippedReads read exceptions - * @param skippedInputs items and corresponding exceptions skipped in - * processing phase - * @param skippedOutputs items and corresponding exceptions skipped in write - * phase - */ - protected void callSkipListeners(final List skippedReads, final Map skippedInputs, - final Map skippedOutputs) { - - for (Exception e : skippedReads) { - try { - listener.onSkipInRead(e); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); - } - } - for (Entry skip : skippedInputs.entrySet()) { - try { - listener.onSkipInProcess(skip.getKey(), skip.getValue()); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skip.getValue()); - } - } - - for (Entry skip : skippedOutputs.entrySet()) { - try { - listener.onSkipInWrite(skip.getKey(), skip.getValue()); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue()); - } - } - } - - /** - * Return a list stored in the attributes under the key. Create an empty - * list and store it if the list is not stored yet. - */ - protected static List getBufferedList(AttributeAccessor attributes, String key) { - List buffer; - if (!attributes.hasAttribute(key)) { - buffer = new ArrayList(); - attributes.setAttribute(key, buffer); - } - else { - @SuppressWarnings("unchecked") - List casted = (List) attributes.getAttribute(key); - buffer = casted; - } - return buffer; - } - - /** - * Return a map of items to exceptions stored in the attributes under the - * key, Create an empty map and store it if the list is not stored yet. - */ - protected static Map getBufferedSkips(AttributeAccessor attributes, String key) { - Map buffer; - if (!attributes.hasAttribute(key)) { - buffer = new LinkedHashMap(); - attributes.setAttribute(key, buffer); - } - else { - @SuppressWarnings("unchecked") - Map casted = (Map) attributes.getAttribute(key); - buffer = casted; - } - return buffer; - } - - /** - * Incorporate retry into the item processor stage. If item processor - * returns null for an input item, it is considered filtered and is not - * added to outputs. - * - * @param inputs the items to process - * @param outputs the items to write - * @param contribution current context - */ - protected void process(final StepContribution contribution, final List inputs, final List outputs, - final Map skippedInputs) throws Exception { - - int filtered = 0; - - for (final I item : inputs) { - - RetryCallback retryCallback = new RetryCallback() { - - public O doWithRetry(RetryContext context) throws Exception { - O output = doProcess(item); - return output; - } - - }; - - RecoveryCallback recoveryCallback = new RecoveryCallback() { - - public O recover(RetryContext context) throws Exception { - Exception e = (Exception) context.getLastThrowable(); - if (processSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { - contribution.incrementProcessSkipCount(); - skippedInputs.put(item, e); - logger.debug("Skipping after failed process", e); - return null; - } - else { - throw new RetryException("Non-skippable exception in recoverer while processing", e); - } - } - - }; - - O output = retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(item, - rollbackClassifier)); - if (output != null) { - outputs.add(output); - } - else { - filtered++; - } - - } - - contribution.incrementFilterCount(filtered); - - } - - /** - * Execute the business logic, delegating to the writer.
- * - * Process the items with the {@link ItemWriter} in a stateful retry. Any - * {@link SkipListener} provided is called when retry attempts are - * exhausted. The listener callback (on write failure) will happen in the - * next transaction automatically.
- */ - protected void write(final List chunk, final StepContribution contribution, final Map skipped) - throws Exception { - - RetryCallback retryCallback = new RetryCallback() { - public Object doWithRetry(RetryContext context) throws Exception { - doWrite(chunk); - contribution.incrementWriteCount(chunk.size()); - return null; - } - }; - - RecoveryCallback recoveryCallback = new RecoveryCallback() { - - public Object recover(RetryContext context) throws Exception { - Exception le = (Exception) context.getLastThrowable(); - if (!writeSkipPolicy.shouldSkip(le, contribution.getSkipCount())) { - throw new RetryException("Non-skippable exception in recoverer", le); - } - if (chunk.size() == 1) { - O item = chunk.get(0); - checkSkipPolicy(item, le, contribution); - return null; - } - if (!rollbackClassifier.classify(le)) { - throw new RetryException( - "Invalid retry state during write caused by exception that does not classify for rollback: ", - le); - } - for (O item : chunk) { - try { - doWrite(Collections.singletonList(item)); - contribution.incrementWriteCount(1); - } - catch (Exception e) { - checkSkipPolicy(item, e, contribution); - if (rollbackClassifier.classify(e)) { - throw e; - } - else { - throw new RetryException( - "Invalid retry state during recovery caused by exception that does not classify for rollback: ", - e); - } - } - } - - return null; - - } - - private void checkSkipPolicy(O item, Exception e, StepContribution contribution) { - if (writeSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { - contribution.incrementWriteSkipCount(); - skipped.put(item, e); - logger.debug("Skipping after failed write", e); - } - else { - throw new RetryException("Non-skippable exception in recoverer", e); - } - } - - }; - - retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(chunk, rollbackClassifier)); - - } -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java deleted file mode 100644 index d08480ad6..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java +++ /dev/null @@ -1,117 +0,0 @@ -package org.springframework.batch.core.step.item; - -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.listener.MulticasterBatchListener; -import org.springframework.batch.core.step.tasklet.Tasklet; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; - -/** - * Superclass for {@link Tasklet}s implementing variations on read-process-write - * item handling. Encapsulates listener registration and bundles listener - * callbacks with relevant method calls. - * - * @author Robert Kasanicky - * - * @param input item type - * @param output item type - */ -public abstract class AbstractItemOrientedTasklet implements Tasklet { - - protected final Log logger = LogFactory.getLog(getClass()); - - protected final ItemReader itemReader; - - protected final ItemProcessor itemProcessor; - - protected final ItemWriter itemWriter; - - protected final MulticasterBatchListener listener = new MulticasterBatchListener(); - - public AbstractItemOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter) { - this.itemReader = itemReader; - this.itemProcessor = itemProcessor; - this.itemWriter = itemWriter; - } - - /** - * Register some {@link StepListener}s with the handler. Each will get the - * callbacks in the order specified at the correct stage. - * - * @param listeners - */ - public void setListeners(StepListener[] listeners) { - for (StepListener listener : listeners) { - registerListener(listener); - } - } - - /** - * Register a listener for callbacks at the appropriate stages in a process. - * - * @param listener a {@link StepListener} - */ - public void registerListener(StepListener listener) { - this.listener.register(listener); - } - - /** - * Surrounds the read call with listener callbacks. - * @return item - * @throws Exception - */ - protected final I doRead() throws Exception { - try { - listener.beforeRead(); - I item = itemReader.read(); - listener.afterRead(item); - return item; - } - catch (Exception e) { - listener.onReadError(e); - throw e; - } - } - - /** - * @param item the input item - * @return the result of the processing - * @throws Exception - */ - protected final O doProcess(I item) throws Exception { - try { - listener.beforeProcess(item); - O result = itemProcessor.process(item); - listener.afterProcess(item, result); - return result; - } - catch (Exception e) { - listener.onProcessError(item, e); - throw e; - } - } - - /** - * Surrounds the actual write call with listener callbacks. - * @param items - * @throws Exception - */ - protected final void doWrite(List items) throws Exception { - try { - listener.beforeWrite(items); - itemWriter.write(items); - listener.afterWrite(items); - } - catch (Exception e) { - listener.onWriteError(e, items); - throw e; - } - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java index 18df6e832..1f9f0a7e9 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java @@ -19,7 +19,6 @@ import java.util.ArrayList; import java.util.List; import org.springframework.batch.core.ChunkListener; -import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.listener.CompositeChunkListener; import org.springframework.batch.repeat.RepeatContext; @@ -79,18 +78,17 @@ abstract class BatchListenerFactoryHelper { } - /** - * @param listeners - */ - public static StepExecutionListener[] getStepListeners(StepListener[] listeners) { - List list = new ArrayList(); + public static List getListeners(StepListener[] listeners, Class cls) { + List list = new ArrayList(); for (int i = 0; i < listeners.length; i++) { - StepListener listener = listeners[i]; - if (listener instanceof StepExecutionListener) { - list.add((StepExecutionListener) listener); + StepListener stepListener = listeners[i]; + if (cls.isAssignableFrom(stepListener.getClass())) { + @SuppressWarnings("unchecked") + T listener = (T) stepListener; + list.add(listener); } } - return list.toArray(new StepExecutionListener[list.size()]); + return list; } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java new file mode 100644 index 000000000..d82774301 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java @@ -0,0 +1,254 @@ +package org.springframework.batch.core.step.item; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; + +import org.springframework.batch.retry.ExhaustedRetryException; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryListener; +import org.springframework.batch.retry.RetryOperations; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.RetryState; +import org.springframework.batch.retry.backoff.BackOffPolicy; +import org.springframework.batch.retry.context.RetryContextSupport; +import org.springframework.batch.retry.policy.RetryContextCache; +import org.springframework.batch.retry.support.DefaultRetryState; +import org.springframework.batch.retry.support.RetrySynchronizationManager; +import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.batch.support.Classifier; + +/** + * A special purpose retry template that deals specifically with multi-valued + * stateful retry. This is useful in the case where the operation to be retried + * operates on multiple items, and when it fails there is no way to decide which + * (if any) of the items was responsible. The {@link RetryState} used in the + * execute methods is composite, and when a failure occurs, all of the keys in + * the composite are "tarred with the same brush". Subsequent attempts to + * execute with any of the keys that have failed previously results in a new + * attempt and the previous state is used to check the {@link RetryPolicy}. If + * one of the failed items eventually succeeds then the others in the current + * composite for that attempt will be cleared from the context cache (as + * normal), but there may still be entries in the cache for the original failed + * items. This might mean that an item that did not cause a failure is never + * retried because other items in the same batch fail fatally first. + * + * @author Dave Syer + * + */ +public class BatchRetryTemplate implements RetryOperations { + + private class BatchRetryState extends DefaultRetryState { + + private final Collection keys; + + public BatchRetryState(Collection keys) { + super(keys); + this.keys = new ArrayList(keys); + } + + } + + private static class BatchRetryContext extends RetryContextSupport { + + private final Collection contexts; + + public BatchRetryContext(RetryContext parent, Collection contexts) { + + super(parent); + + this.contexts = contexts; + int count = 0; + + for (RetryContext context : contexts) { + int retryCount = context.getRetryCount(); + if (retryCount > count) { + count = retryCount; + registerThrowable(context.getLastThrowable()); + } + } + + } + + } + + private static class InnerRetryTemplate extends RetryTemplate { + + @Override + protected boolean canRetry(RetryPolicy retryPolicy, RetryContext context) { + + BatchRetryContext batchContext = (BatchRetryContext) context; + + for (RetryContext nextContext : batchContext.contexts) { + if (!super.canRetry(retryPolicy, nextContext)) { + return false; + } + } + + return true; + + } + + @Override + protected RetryContext open(RetryPolicy retryPolicy, RetryState state) { + + BatchRetryState batchState = (BatchRetryState) state; + + Collection contexts = new ArrayList(); + for (RetryState retryState : batchState.keys) { + contexts.add(super.open(retryPolicy, retryState)); + } + + return new BatchRetryContext(RetrySynchronizationManager.getContext(), contexts); + + } + + @Override + protected void registerThrowable(RetryPolicy retryPolicy, RetryState state, RetryContext context, Exception e) { + + BatchRetryState batchState = (BatchRetryState) state; + BatchRetryContext batchContext = (BatchRetryContext) context; + + Iterator contextIterator = batchContext.contexts.iterator(); + for (RetryState retryState : batchState.keys) { + RetryContext nextContext = contextIterator.next(); + super.registerThrowable(retryPolicy, retryState, nextContext, e); + } + + } + + @Override + protected void close(RetryPolicy retryPolicy, RetryContext context, RetryState state, boolean succeeded) { + + BatchRetryState batchState = (BatchRetryState) state; + BatchRetryContext batchContext = (BatchRetryContext) context; + + Iterator contextIterator = batchContext.contexts.iterator(); + for (RetryState retryState : batchState.keys) { + RetryContext nextContext = contextIterator.next(); + super.close(retryPolicy, nextContext, retryState, succeeded); + } + + } + + @Override + protected T handleRetryExhausted(RecoveryCallback recoveryCallback, RetryContext context, + RetryState state) throws Exception { + + BatchRetryState batchState = (BatchRetryState) state; + BatchRetryContext batchContext = (BatchRetryContext) context; + + // Accumulate exceptions to be thrown so all the keys get a crack + Exception rethrowable = null; + ExhaustedRetryException exhausted = null; + + Iterator contextIterator = batchContext.contexts.iterator(); + for (RetryState retryState : batchState.keys) { + + RetryContext nextContext = contextIterator.next(); + + try { + super.handleRetryExhausted(null, nextContext, retryState); + } + catch (ExhaustedRetryException e) { + exhausted = e; + } + catch (Exception e) { + rethrowable = e; + } + + } + + if (recoveryCallback != null) { + return recoveryCallback.recover(context); + } + + if (exhausted != null) { + throw exhausted; + } + + throw rethrowable; + + } + + } + + private final InnerRetryTemplate delegate = new InnerRetryTemplate(); + + private final RetryTemplate regular = new RetryTemplate(); + + public T execute(RetryCallback retryCallback, Collection states) throws ExhaustedRetryException, + Exception { + RetryState batchState = new BatchRetryState(states); + return delegate.execute(retryCallback, batchState); + } + + public T execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, + Collection states) throws ExhaustedRetryException, Exception { + RetryState batchState = new BatchRetryState(states); + return delegate.execute(retryCallback, recoveryCallback, batchState); + } + + public final T execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, + RetryState retryState) throws Exception, ExhaustedRetryException { + return regular.execute(retryCallback, recoveryCallback, retryState); + } + + public final T execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback) throws Exception { + return regular.execute(retryCallback, recoveryCallback); + } + + public final T execute(RetryCallback retryCallback, RetryState retryState) throws Exception, + ExhaustedRetryException { + return regular.execute(retryCallback, retryState); + } + + public final T execute(RetryCallback retryCallback) throws Exception { + return regular.execute(retryCallback); + } + + public static List createState(List keys) { + List states = new ArrayList(); + for (Object key : keys) { + states.add(new DefaultRetryState(key)); + } + return states; + } + + public static List createState(List keys, Classifier classifier) { + List states = new ArrayList(); + for (Object key : keys) { + states.add(new DefaultRetryState(key, classifier)); + } + return states; + } + + public void registerListener(RetryListener listener) { + delegate.registerListener(listener); + regular.registerListener(listener); + } + + public void setBackOffPolicy(BackOffPolicy backOffPolicy) { + delegate.setBackOffPolicy(backOffPolicy); + regular.setBackOffPolicy(backOffPolicy); + } + + public void setListeners(RetryListener[] listeners) { + delegate.setListeners(listeners); + regular.setListeners(listeners); + } + + public void setRetryContextCache(RetryContextCache retryContextCache) { + delegate.setRetryContextCache(retryContextCache); + regular.setRetryContextCache(retryContextCache); + } + + public void setRetryPolicy(RetryPolicy retryPolicy) { + delegate.setRetryPolicy(retryPolicy); + regular.setRetryPolicy(retryPolicy); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java index b0e510fa3..444d529e7 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java @@ -9,8 +9,8 @@ import java.util.List; * Encapsulation of a list of items to be processed and possibly a list of * failed items to be skipped. To mark an item as skipped clients should iterate * over the chunk using the {@link #iterator()} method, and if there is a - * failure call {@link ChunkIterator#remove(Exception)} on the iterator. The - * skipped items are then available through the chunk. + * failure call {@link ChunkIterator#remove(Exception)} on the iterator. + * The skipped items are then available through the chunk. * * @author Dave Syer * @@ -19,7 +19,27 @@ class Chunk implements Iterable { private List items = new ArrayList(); - private List> skips = new ArrayList>(); + private List> skips = new ArrayList>(); + + private List errors = new ArrayList(); + + private Object userData; + + private boolean end; + + public Chunk() { + this(null,null); + } + + public Chunk(List items, List> skips) { + super(); + if (items!=null) { + this.items = new ArrayList(items); + } + if (skips!=null) { + this.skips = new ArrayList>(skips); + } + } /** * Add the item to the chunk. @@ -34,6 +54,8 @@ class Chunk implements Iterable { */ public void clear() { items.clear(); + skips.clear(); + userData = null; } /** @@ -46,8 +68,25 @@ class Chunk implements Iterable { /** * @return a copy of the skips as an unmodifiable list */ - public List> getSkips() { - return Collections.unmodifiableList(new ArrayList>(skips)); + public List> getSkips() { + return Collections.unmodifiableList(skips); + } + + /** + * @return a copy of the anonymous errros as an unmodifiable list + */ + public List getErrors() { + return Collections.unmodifiableList(errors); + } + + /** + * Register an anonymous skip. To skip an individual item, use + * {@link ChunkIterator#remove()}. + * + * @param e the exception that caused the skip + */ + public void skip(Exception e) { + errors.add(e); } /** @@ -72,6 +111,22 @@ class Chunk implements Iterable { return items.size(); } + public boolean isEnd() { + return end; + } + + public void setEnd() { + this.end = true; + } + + public Object getUserData() { + return userData; + } + + public void setUserData(Object userData) { + this.userData = userData; + } + /* * (non-Javadoc) * @@ -83,8 +138,9 @@ class Chunk implements Iterable { } /** - * Special iterator for a chunk providing the {@link #remove(Exception)} - * method for dynamically removing an item abd adding it to the skips. + * Special iterator for a chunk providing the + * {@link #remove(Exception)} method for dynamically removing an + * item and adding it to the skips. * * @author Dave Syer * @@ -109,6 +165,11 @@ class Chunk implements Iterable { } public void remove(Exception e) { + remove(); + skips.add(new SkipWrapper(next, e)); + } + + public void remove() { if (next == null) { if (iterator.hasNext()) { next = iterator.next(); @@ -117,14 +178,9 @@ class Chunk implements Iterable { return; } } - skips.add(new ItemWrapper(next, e)); iterator.remove(); } - public void remove() { - throw new UnsupportedOperationException("To remove an item you must provide an exception."); - } - } } \ No newline at end of file diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java new file mode 100644 index 000000000..1fcd5bcb8 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java @@ -0,0 +1,67 @@ +package org.springframework.batch.core.step.item; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.repeat.RepeatStatus; +import org.springframework.core.AttributeAccessor; + +/** + * A {@link Tasklet} implementing variations on read-process-write item + * handling. + * + * @author Dave Syer + * + * @param input item type + */ +public class ChunkOrientedTasklet implements Tasklet { + + private static final String INPUTS_KEY = "INPUTS"; + + private final ChunkProcessor chunkProcessor; + + private final ChunkProvider chunkProvider; + + private boolean buffering = true; + + public ChunkOrientedTasklet(ChunkProvider chunkProvider, ChunkProcessor chunkProcessor) { + this.chunkProvider = chunkProvider; + this.chunkProcessor = chunkProcessor; + } + + /** + * Flag to indicate that items should be buffered once read. Defaults to + * true, which is appropriate for forward-only, non-transactional item + * readers. Main (or only) use case for setting this flag to true is a + * transactional JMS item reader. + * + * @param buffering + */ + public void setBuffering(boolean buffering) { + this.buffering = buffering; + } + + public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception { + + @SuppressWarnings("unchecked") + Chunk inputs = (Chunk) attributes.getAttribute(INPUTS_KEY); + if (inputs == null) { + inputs = chunkProvider.provide(contribution); + if (buffering) { + attributes.setAttribute(INPUTS_KEY, inputs); + } + } + + chunkProcessor.process(contribution, inputs); + + attributes.removeAttribute(INPUTS_KEY); + chunkProvider.postProcess(contribution, inputs); + if (!inputs.isEnd()) { + contribution.setExitStatus(ExitStatus.FINISHED); + } + + return RepeatStatus.continueIf(!inputs.isEnd()); + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProcessor.java new file mode 100644 index 000000000..0dba281da --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProcessor.java @@ -0,0 +1,9 @@ +package org.springframework.batch.core.step.item; + +import org.springframework.batch.core.StepContribution; + +public interface ChunkProcessor { + + void process(StepContribution contribution, Chunk chunk) throws Exception; + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProvider.java new file mode 100644 index 000000000..97c3cfaf2 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProvider.java @@ -0,0 +1,11 @@ +package org.springframework.batch.core.step.item; + +import org.springframework.batch.core.StepContribution; + +public interface ChunkProvider { + + Chunk provide(StepContribution contribution) throws Exception; + + void postProcess(StepContribution contribution, Chunk chunk); + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java deleted file mode 100644 index 8e18bab10..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java +++ /dev/null @@ -1,162 +0,0 @@ -/* - * 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.core.step.item; - -import java.util.ArrayList; -import java.util.List; -import java.util.Map; - -import org.springframework.batch.core.ExitStatus; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.skip.SkipPolicy; -import org.springframework.batch.core.step.skip.NonSkippableReadException; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.RepeatStatus; -import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.support.Classifier; -import org.springframework.core.AttributeAccessor; - -/** - * If there is an exception on input it is skipped if allowed. If there is an - * exception on output, it will be re-thrown in any case, and the behaviour when - * the item is next encountered depends on the retryable and skippable exception - * configuration. If the exception is retryable the write will be attempted - * again up to the retry limit. When retry attempts are exhausted the skip - * listener is invoked and the skip count incremented. A retryable exception is - * thus also effectively also implicitly skippable. - * - * ItemProcessor is assumed to be transactional. In case of - * rollback caused by error on write the processing phase will be repeated. - * - * @param input item type - * @param output item type - * - * @author Dave Syer - * @author Robert Kasanicky - */ -public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolerantChunkOrientedTasklet { - - final static private String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY"; - - public FaultTolerantChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, - RepeatOperations chunkOperations, RetryOperations retryTemplate, - Classifier rollbackClassifier, SkipPolicy readSkipPolicy, - SkipPolicy writeSkipPolicy, SkipPolicy processSkipPolicy) { - - super(itemReader, itemProcessor, itemWriter, retryTemplate, readSkipPolicy, processSkipPolicy, writeSkipPolicy, - rollbackClassifier, chunkOperations); - } - - /** - * Read the next chunk of items and if not empty pass the items one-by-one - * to {@link #process(StepContribution, List, List, Map)} and finally write - * all items by {@link #write(List, StepContribution, Map)}. - * - * @see org.springframework.batch.core.step.tasklet.Tasklet#execute(org.springframework.batch.core.StepContribution, - * AttributeAccessor) - */ - public RepeatStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - - final List inputs = getBufferedList(attributes, INPUT_BUFFER_KEY); - final List outputs = new ArrayList(); - - final List skippedReads = getBufferedList(attributes, SKIPPED_READS_KEY); - - // TODO: invert logic below so that default can be FINISHED? - RepeatStatus continuable = RepeatStatus.CONTINUABLE; - - if (inputs.isEmpty() && outputs.isEmpty()) { - - continuable = getRepeatOperations().iterate(new RepeatCallback() { - public RepeatStatus doInIteration(final RepeatContext context) throws Exception { - I item = read(contribution, skippedReads); - - if (item == null) { - return RepeatStatus.FINISHED; - } - inputs.add(item); - contribution.incrementReadCount(); - return RepeatStatus.CONTINUABLE; - } - }); - - ExitStatus status = continuable.isContinuable() ? ExitStatus.EXECUTING : ExitStatus.FINISHED; - contribution.setExitStatus(status); - - } - - final Map skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY); - final Map skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY); - - if (!inputs.isEmpty()) { - inputs.removeAll(skippedInputs.keySet()); - process(contribution, inputs, outputs, skippedInputs); - - outputs.removeAll(skippedOutputs.keySet()); - write(outputs, contribution, skippedOutputs); - } - - callSkipListeners(skippedReads, skippedInputs, skippedOutputs); - - // On successful completion clear the attributes to signal that there is - // no more processing - for (String key : attributes.attributeNames()) { - attributes.removeAttribute(key); - } - - return continuable; - - } - - /** - * Tries to read the item from the reader, in case of exception skip the - * item if the skip policy allows, otherwise re-throw. - * - * @param contribution current StepContribution holding skipped items count - * @param skippedReads - * @return next item for processing - */ - protected I read(StepContribution contribution, List skippedReads) throws Exception { - - while (true) { - try { - return doRead(); - } - catch (Exception e) { - - if (getReadSkipPolicy().shouldSkip(e, contribution.getStepSkipCount())) { - // increment skip count and try again - contribution.incrementReadSkipCount(); - skippedReads.add(e); - - logger.debug("Skipping failed input", e); - } - else { - throw new NonSkippableReadException("Non-skippable exception during read", e); - } - - } - } - - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java new file mode 100644 index 000000000..9382ae133 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -0,0 +1,220 @@ +package org.springframework.batch.core.step.item; + +import java.util.Collections; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; +import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryException; +import org.springframework.batch.retry.support.DefaultRetryState; +import org.springframework.batch.support.Classifier; + +public class FaultTolerantChunkProcessor extends SimpleChunkProcessor { + + private SkipPolicy itemProcessSkipPolicy = new LimitCheckingItemSkipPolicy(0); + + private SkipPolicy itemWriteSkipPolicy = new LimitCheckingItemSkipPolicy(0); + + private final BatchRetryTemplate batchRetryTemplate; + + private Classifier rollbackClassifier; + + private Log logger = LogFactory.getLog(getClass()); + + private boolean buffering; + + public void setProcessSkipPolicy(SkipPolicy SkipPolicy) { + this.itemProcessSkipPolicy = SkipPolicy; + } + + public void setWriteSkipPolicy(SkipPolicy SkipPolicy) { + this.itemWriteSkipPolicy = SkipPolicy; + } + + public void setRollbackClassifier(Classifier rollbackClassifier) { + this.rollbackClassifier = rollbackClassifier; + } + + public void setBuffering(boolean buffering) { + this.buffering = buffering; + } + + public FaultTolerantChunkProcessor(ItemProcessor itemProcessor, + ItemWriter itemWriter, BatchRetryTemplate batchRetryTemplate) { + super(itemProcessor, itemWriter); + this.batchRetryTemplate = batchRetryTemplate; + } + + @Override + protected Chunk transform(final StepContribution contribution, Chunk inputs) throws Exception { + + Chunk outputs = new Chunk(); + + for (final Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { + + final I item = iterator.next(); + + RetryCallback retryCallback = new RetryCallback() { + + public O doWithRetry(RetryContext context) throws Exception { + O output = doProcess(item); + if (output == null) { + // No need to re-process filtered items + iterator.remove(); + } + return output; + } + + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public O recover(RetryContext context) throws Exception { + Exception e = (Exception) context.getLastThrowable(); + if (itemProcessSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementProcessSkipCount(); + iterator.remove(e); + logger.debug("Skipping after failed process", e); + return null; + } + else { + throw new RetryException("Non-skippable exception in recoverer while processing", e); + } + } + + }; + + // TODO: is it OK to use the item as a key for the retry state? + O output = batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(item, + rollbackClassifier)); + if (output != null) { + outputs.add(output); + } + + } + + return outputs; + + } + + @Override + protected void write(final StepContribution contribution, final Chunk inputs, final Chunk outputs) + throws Exception { + + RetryCallback retryCallback = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Exception { + doWrite(outputs.getItems()); + contribution.incrementWriteCount(outputs.size()); + return null; + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + + public Object recover(RetryContext context) throws Exception { + + Exception le = (Exception) context.getLastThrowable(); + if (outputs.size() > 1 && !rollbackClassifier.classify(le)) { + throw new RetryException("Invalid retry state during write caused by " + + "exception that does not classify for rollback: ", le); + } + + boolean singleton = outputs.size() == 1; + + Chunk.ChunkIterator inputIterator = inputs.iterator(); + for (Chunk.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) { + + inputIterator.next(); + O item = outputIterator.next(); + if (singleton) { + checkSkipPolicy(inputIterator, outputIterator, le, contribution); + return null; + } + + try { + doWrite(Collections.singletonList(item)); + contribution.incrementWriteCount(1); + } + catch (Exception e) { + checkSkipPolicy(inputIterator, outputIterator, e, contribution); + if (rollbackClassifier.classify(e)) { + throw e; + } + else { + throw new RetryException( + "Invalid retry state during recovery caused by exception that does not classify for rollback: ", + e); + } + } + + } + + return null; + + } + + }; + + RecoveryCallback batchRecoveryCallback = new RecoveryCallback() { + + public Object recover(RetryContext context) throws Exception { + + Exception e = (Exception) context.getLastThrowable(); + if (outputs.size() > 1 && !rollbackClassifier.classify(e)) { + throw new RetryException("Invalid retry state during write caused by " + + "exception that does not classify for rollback: ", e); + } + + Chunk.ChunkIterator inputIterator = inputs.iterator(); + for (Chunk.ChunkIterator outputIterator = outputs.iterator(); outputIterator.hasNext();) { + + inputIterator.next(); + outputIterator.next(); + + checkSkipPolicy(inputIterator, outputIterator, e, contribution); + if (!rollbackClassifier.classify(e)) { + throw new RetryException( + "Invalid retry state during recovery caused by exception that does not classify for rollback: ", + e); + } + + } + + return null; + + } + + }; + + if (!buffering) { + batchRetryTemplate.execute(retryCallback, batchRecoveryCallback, BatchRetryTemplate.createState(inputs + .getItems(), rollbackClassifier)); + } + else { + batchRetryTemplate.execute(retryCallback, recoveryCallback, new DefaultRetryState(inputs, + rollbackClassifier)); + } + + } + + private void checkSkipPolicy(Chunk.ChunkIterator inputIterator, Chunk.ChunkIterator outputIterator, + Exception e, StepContribution contribution) { + if (itemWriteSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + contribution.incrementWriteSkipCount(); + inputIterator.remove(); + outputIterator.remove(e); + logger.debug("Skipping after failed write", e); + } + else { + throw new RetryException("Non-skippable exception in recoverer", e); + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java new file mode 100644 index 000000000..0b786bcc9 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java @@ -0,0 +1,45 @@ +package org.springframework.batch.core.step.item; + +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; +import org.springframework.batch.core.step.skip.NonSkippableReadException; +import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.repeat.RepeatOperations; + +public class FaultTolerantChunkProvider extends SimpleChunkProvider { + + private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(0); + + public FaultTolerantChunkProvider(ItemReader itemReader, RepeatOperations repeatOperations) { + super(itemReader, repeatOperations); + } + + public void setSkipPolicy(SkipPolicy SkipPolicy) { + this.skipPolicy = SkipPolicy; + } + + @Override + protected I read(StepContribution contribution, Chunk chunk) throws Exception { + while (true) { + try { + return doRead(); + } + catch (Exception e) { + + if (skipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + // increment skip count and try again + contribution.incrementReadSkipCount(); + chunk.skip(e); + + logger.debug("Skipping failed input", e); + } + else { + throw new NonSkippableReadException("Non-skippable exception during read", e); + } + + } + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java index bfebc9346..effa154e6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java @@ -6,11 +6,16 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; -import org.springframework.batch.core.step.skip.SkipPolicy; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.step.item.SimpleRetryExceptionHandler; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableReadException; import org.springframework.batch.core.step.skip.SkipLimitExceededException; import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.core.step.skip.SkipPolicy; import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; @@ -22,7 +27,6 @@ import org.springframework.batch.retry.policy.ExceptionClassifierRetryPolicy; import org.springframework.batch.retry.policy.MapRetryContextCache; import org.springframework.batch.retry.policy.RetryContextCache; import org.springframework.batch.retry.policy.SimpleRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.batch.support.Classifier; /** @@ -159,12 +163,6 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean rollbackClassifier = new Classifier() { - public Boolean classify(Throwable classifiable) { - return getTransactionAttribute().rollbackOn(classifiable); - } - }; + batchRetryTemplate.setRetryPolicy(retryPolicy); // Co-ordinate the retry policy with the exception handler: RepeatOperations stepOperations = getStepOperations(); if (stepOperations instanceof RepeatTemplate) { - ((RepeatTemplate) stepOperations).setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, - getExceptionHandler(), fatalExceptionClasses)); + SimpleRetryExceptionHandler exceptionHandler = new SimpleRetryExceptionHandler(retryPolicy, + getExceptionHandler(), fatalExceptionClasses); + ((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler); } if (retryContextCache == null) { if (cacheCapacity > 0) { - retryTemplate.setRetryContextCache(new MapRetryContextCache(cacheCapacity)); + batchRetryTemplate.setRetryContextCache(new MapRetryContextCache(cacheCapacity)); } } else { - retryTemplate.setRetryContextCache(retryContextCache); + batchRetryTemplate.setRetryContextCache(retryContextCache); } if (retryListeners != null) { - retryTemplate.setListeners(retryListeners); + batchRetryTemplate.setListeners(retryListeners); } List> exceptions = new ArrayList>( @@ -262,23 +255,32 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean>(retryableExceptionClasses)); SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, new ArrayList>(fatalExceptionClasses)); + + Classifier rollbackClassifier = new Classifier() { + public Boolean classify(Throwable classifiable) { + return getTransactionAttribute().rollbackOn(classifiable); + } + }; - if (isReaderTransactionalQueue) { - NonbufferingFaultTolerantChunkOrientedTasklet tasklet = new NonbufferingFaultTolerantChunkOrientedTasklet( - getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, - rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - tasklet.setListeners(getListeners()); + FaultTolerantChunkProvider chunkProvider = new FaultTolerantChunkProvider(getItemReader(), + getChunkOperations()); + chunkProvider.setSkipPolicy(readSkipPolicy); + chunkProvider.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemReadListener.class)); + chunkProvider.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), SkipListener.class)); - step.setTasklet(tasklet); - } - else { - FaultTolerantChunkOrientedTasklet tasklet = new FaultTolerantChunkOrientedTasklet( - getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, - rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - tasklet.setListeners(getListeners()); + FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor(getItemProcessor(), getItemWriter(), batchRetryTemplate); + chunkProcessor.setBuffering(!isReaderTransactionalQueue); + chunkProcessor.setWriteSkipPolicy(writeSkipPolicy); + chunkProcessor.setProcessSkipPolicy(writeSkipPolicy); + chunkProcessor.setRollbackClassifier(rollbackClassifier); + chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemProcessListener.class)); + chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemWriteListener.class)); + chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), SkipListener.class)); - step.setTasklet(tasklet); - } + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + tasklet.setBuffering(!isReaderTransactionalQueue); + + step.setTasklet(tasklet); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java deleted file mode 100644 index 2b32a8b52..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java +++ /dev/null @@ -1,174 +0,0 @@ -package org.springframework.batch.core.step.item; - -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - -import org.springframework.batch.core.ExitStatus; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.skip.SkipPolicy; -import org.springframework.batch.core.step.skip.SkipListenerFailedException; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.RepeatStatus; -import org.springframework.batch.retry.RetryOperations; -import org.springframework.batch.support.Classifier; -import org.springframework.core.AttributeAccessor; - -/** - * Fault-tolerant chunk-oriented tasklet implementation which does not buffer - * items that have been read - the assumption is that item reader is - * transactional and will re-present the items after transaction rollback, while - * item ordering might not be preserved (JMS). - * - * Note that the implementation relies on {@link Object#equals(Object)} - * comparisons for recognizing items on retry/skip. - * - * @param input item type - * @param output item type - * - * @author Robert Kasanicky - */ -public class NonbufferingFaultTolerantChunkOrientedTasklet extends - AbstractFaultTolerantChunkOrientedTasklet { - - public NonbufferingFaultTolerantChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, - RepeatOperations chunkOperations, RetryOperations retryTemplate, - Classifier rollbackClassifier, SkipPolicy readSkipPolicy, - SkipPolicy writeSkipPolicy, SkipPolicy processSkipPolicy) { - - super(itemReader, itemProcessor, itemWriter, retryTemplate, readSkipPolicy, processSkipPolicy, writeSkipPolicy, - rollbackClassifier, chunkOperations); - } - - /** - * Read-process-write a list of items. Uses fault-tolerant read, process and - * write implementations. - */ - public RepeatStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - final List inputs = new ArrayList(); - - final List skippedReads = getBufferedList(attributes, SKIPPED_READS_KEY); - RepeatStatus continuable = getRepeatOperations().iterate(new RepeatCallback() { - public RepeatStatus doInIteration(final RepeatContext context) throws Exception { - I item = read(contribution, skippedReads); - - if (item == null) { - return RepeatStatus.FINISHED; - } - inputs.add(item); - contribution.incrementReadCount(); - return RepeatStatus.CONTINUABLE; - } - }); - - ExitStatus result = continuable.isContinuable() ? ExitStatus.EXECUTING : ExitStatus.FINISHED; - contribution.setExitStatus(result); - - // filter inputs marked for skipping - final Map skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY); - final Map skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY); - final Set inputsIncludingSkips = new HashSet(inputs.size()); - final Set outputsIncludingSkips = new HashSet(inputs.size()); - - if (!inputs.isEmpty()) { - inputsIncludingSkips.addAll(inputs); - inputs.removeAll(skippedInputs.keySet()); - - final List outputs = new ArrayList(); - process(contribution, inputs, outputs, skippedInputs); - - // filter outputs marked for skipping - outputsIncludingSkips.addAll(outputs); - outputs.removeAll(skippedOutputs.keySet()); - - write(outputs, contribution, skippedOutputs); - } - - callSkipListenersAndCleanSkipsFromBuffer(skippedReads, skippedInputs, skippedOutputs, inputsIncludingSkips, - outputsIncludingSkips); - - return continuable; - } - - /** - * Identify items successfully skipped in this tasklet iteration, call skip - * listeners and remove skips from buffer. This requires care, because we - * might be processing a different chunk after rollback i.e. items marked - * for skipping from previous tasklet iteration may not have been - * encountered now. - */ - private void callSkipListenersAndCleanSkipsFromBuffer(final List skippedReads, - final Map skippedInputs, final Map skippedOutputs, - final Set inputsIncludingSkips, final Set outputsIncludingSkips) { - for (Exception skippedReadException : skippedReads) { - try { - listener.onSkipInRead(skippedReadException); - } - catch (RuntimeException e) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, skippedReadException); - } - } - skippedReads.clear(); - for (I input : inputsIncludingSkips) { - if (skippedInputs.containsKey(input)) { - try { - listener.onSkipInProcess(input, skippedInputs.get(input)); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skippedInputs - .get(input)); - } - skippedInputs.remove(input); - } - } - for (O output : outputsIncludingSkips) { - if (skippedOutputs.containsKey(output)) { - try { - listener.onSkipInWrite(output, skippedOutputs.get(output)); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skippedOutputs - .get(output)); - } - skippedOutputs.remove(output); - } - } - } - - /** - * Tries to read the item from the reader, in case of exception skip the - * skip listener is called and exception is re-thrown (failed read causes - * rollback automatically because the reader is assumed to be - * transactional). - * - * @param contribution current StepContribution holding skipped items count - * @return next item for processing - */ - protected I read(StepContribution contribution, final List skipped) throws Exception { - - try { - return doRead(); - } - catch (Exception e) { - - if (getReadSkipPolicy().shouldSkip(e, contribution.getStepSkipCount())) { - // increment skip count and try again - contribution.incrementReadSkipCount(); - skipped.add(e); - logger.debug("Skipping failed input", e); - } - - throw e; - } - - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTasklet.java deleted file mode 100644 index 6b98b4ca9..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTasklet.java +++ /dev/null @@ -1,81 +0,0 @@ -package org.springframework.batch.core.step.item; - -import java.util.ArrayList; -import java.util.List; - -import org.springframework.batch.core.ExitStatus; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.step.tasklet.Tasklet; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.RepeatStatus; -import org.springframework.core.AttributeAccessor; - -/** - * Simplest possible implementation of chunk-oriented {@link Tasklet} with no - * skipping or recovering. Just delegates all calls to the provided - * {@link ItemReader}, {@link ItemProcessor} and {@link ItemWriter}. - * - * @author Dave Syer - * @author Robert Kasanicky - */ -public class SimpleChunkOrientedTasklet extends AbstractItemOrientedTasklet { - - private RepeatOperations repeatOperations; - - public SimpleChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, - RepeatOperations repeatOperations) { - super(itemReader, itemProcessor, itemWriter); - this.repeatOperations = repeatOperations; - } - - /** - * Read-process-write a list of items. - */ - public RepeatStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - ExitStatus result = ExitStatus.EXECUTING; - final List inputs = new ArrayList(); - - RepeatStatus continuable = repeatOperations.iterate(new RepeatCallback() { - - public RepeatStatus doInIteration(final RepeatContext context) throws Exception { - I item = doRead(); - - if (item == null) { - return RepeatStatus.FINISHED; - } - inputs.add(item); - contribution.incrementReadCount(); - return RepeatStatus.CONTINUABLE; - } - }); - - result = continuable.isContinuable() ? ExitStatus.EXECUTING : ExitStatus.FINISHED; - contribution.setExitStatus(result); - - // If there is no input we don't have to do anything more - if (inputs.isEmpty()) { - return continuable; - } - - List outputs = new ArrayList(); - for (I item : inputs) { - O output = doProcess(item); - if (output != null) { - outputs.add(output); - } - } - contribution.incrementFilterCount(inputs.size() - outputs.size()); - - doWrite(outputs); - contribution.incrementWriteCount(outputs.size()); - - return continuable; - } - -} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java new file mode 100644 index 000000000..4fb972595 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java @@ -0,0 +1,159 @@ +package org.springframework.batch.core.step.item; + +import java.util.List; + +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.listener.MulticasterBatchListener; +import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemWriter; + +public class SimpleChunkProcessor implements ChunkProcessor { + + private final ItemProcessor itemProcessor; + + private final ItemWriter itemWriter; + + private final MulticasterBatchListener listener = new MulticasterBatchListener(); + + public SimpleChunkProcessor(ItemProcessor itemProcessor, ItemWriter itemWriter) { + this.itemProcessor = itemProcessor; + this.itemWriter = itemWriter; + } + + /** + * Register some {@link StepListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage. + * + * @param listeners + */ + public void setListeners(List listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a process. + * + * @param listener a {@link StepListener} + */ + public void registerListener(StepListener listener) { + this.listener.register(listener); + } + + /** + * @param item the input item + * @return the result of the processing + * @throws Exception + */ + protected final O doProcess(I item) throws Exception { + try { + listener.beforeProcess(item); + O result = itemProcessor.process(item); + listener.afterProcess(item, result); + return result; + } + catch (Exception e) { + listener.onProcessError(item, e); + throw e; + } + } + + /** + * Surrounds the actual write call with listener callbacks. + * + * @param items + * @throws Exception + */ + protected final void doWrite(List items) throws Exception { + try { + listener.beforeWrite(items); + itemWriter.write(items); + listener.afterWrite(items); + } + catch (Exception e) { + listener.onWriteError(e, items); + throw e; + } + } + + public final void process(StepContribution contribution, Chunk inputs) throws Exception { + + // If there is no input we don't have to do anything more + if (inputs.isEmpty()) { + return; + } + + Chunk outputs = transform(contribution, inputs); + + contribution.incrementFilterCount(inputs.size() - outputs.size()); + + /* + * Need to remember the write skips across transactions, otherwise they + * keep coming back. Since we register skips with the inputs they will + * not be processed again but the output skips need to be saved for + * registration later with the listeners. The inputs are going to be the + * same for all transactions processing the same chunk, but the outputs + * are not, so we stash them in user data on the inputs. + */ + + @SuppressWarnings("unchecked") + Chunk skips = (Chunk) inputs.getUserData(); + if (skips == null) { + skips = new Chunk(); + } + + outputs = new Chunk(outputs.getItems(), skips.getSkips()); + inputs.setUserData(outputs); + + write(contribution, inputs, outputs); + + for (SkipWrapper wrapper : inputs.getSkips()) { + I item = wrapper.getItem(); + if (item == null) { + continue; + } + Exception e = wrapper.getException(); + try { + listener.onSkipInProcess(item, e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + } + + for (SkipWrapper wrapper : outputs.getSkips()) { + Exception e = wrapper.getException(); + try { + listener.onSkipInWrite(wrapper.getItem(), e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + } + + } + + protected void write(StepContribution contribution, Chunk inputs, Chunk outputs) throws Exception { + doWrite(outputs.getItems()); + contribution.incrementWriteCount(outputs.size()); + } + + protected Chunk transform(StepContribution contribution, Chunk inputs) throws Exception { + Chunk outputs = new Chunk(); + for (Chunk.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) { + final I item = iterator.next(); + O output = doProcess(item); + if (output != null) { + outputs.add(output); + } + else { + iterator.remove(); + } + } + return outputs; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java new file mode 100644 index 000000000..0e4696c26 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java @@ -0,0 +1,114 @@ +package org.springframework.batch.core.step.item; + +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.listener.MulticasterBatchListener; +import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.repeat.RepeatCallback; +import org.springframework.batch.repeat.RepeatContext; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.RepeatStatus; + +/** + * + * @author Dave Syer + * + * @param input item type + */ +public class SimpleChunkProvider implements ChunkProvider { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final ItemReader itemReader; + + private final MulticasterBatchListener listener = new MulticasterBatchListener(); + + private final RepeatOperations repeatOperations; + + public SimpleChunkProvider(ItemReader itemReader, RepeatOperations repeatOperations) { + this.itemReader = itemReader; + this.repeatOperations = repeatOperations; + } + + /** + * Register some {@link StepListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage. + * + * @param listeners + */ + public void setListeners(List listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a process. + * + * @param listener a {@link StepListener} + */ + public void registerListener(StepListener listener) { + this.listener.register(listener); + } + + /** + * Surrounds the read call with listener callbacks. + * @return item + * @throws Exception + */ + protected final I doRead() throws Exception { + try { + listener.beforeRead(); + I item = itemReader.read(); + listener.afterRead(item); + return item; + } + catch (Exception e) { + listener.onReadError(e); + throw e; + } + } + + public Chunk provide(final StepContribution contribution) throws Exception { + + final Chunk inputs = new Chunk(); + repeatOperations.iterate(new RepeatCallback() { + + public RepeatStatus doInIteration(final RepeatContext context) throws Exception { + I item = read(contribution, inputs); + if (item == null) { + inputs.setEnd(); + return RepeatStatus.FINISHED; + } + inputs.add(item); + contribution.incrementReadCount(); + return RepeatStatus.CONTINUABLE; + } + + }); + + return inputs; + + } + + public void postProcess(StepContribution contribution, Chunk chunk) { + for (Exception e : chunk.getErrors()) { + try { + listener.onSkipInRead(e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + } + } + + protected I read(StepContribution contribution, Chunk chunk) throws Exception { + return doRead(); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 5b92551d6..032001b57 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,8 +15,13 @@ */ package org.springframework.batch.core.step.item; +import java.util.List; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.StepListener; @@ -26,7 +31,6 @@ import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.validator.Validator; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.exception.DefaultExceptionHandler; @@ -43,18 +47,16 @@ import org.springframework.transaction.interceptor.TransactionAttribute; import org.springframework.util.Assert; /** - * Most common configuration options for simple steps should be found here. Use + * Most common configuration options for simple steps should be found here. Use * this factory bean instead of creating a {@link Step} implementation manually. * * This factory does not support configuration of fault-tolerant behavior, use * appropriate subclass of this factory bean to configure skip or retry. * - * @see FaultTolerantStepFactoryBean - * * @author Dave Syer * */ -public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { +public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { private static final int DEFAULT_COMMIT_INTERVAL = 1; @@ -69,15 +71,13 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { private ItemWriter itemWriter; private PlatformTransactionManager transactionManager; - + private TransactionAttribute transactionAttribute; private JobRepository jobRepository; private boolean singleton = true; - private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1); - private ItemStream[] streams = new ItemStream[0]; private StepListener[] listeners = new StepListener[0]; @@ -86,7 +86,9 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { private ItemProcessor itemProcessor = new ItemProcessor() { @SuppressWarnings("unchecked") - public S process(T item) throws Exception {return (S)item;} + public S process(T item) throws Exception { + return (S) item; + } }; private int commitInterval = 0; @@ -253,13 +255,13 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { * @return the transactionAttribute */ protected TransactionAttribute getTransactionAttribute() { - return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute(){ + return transactionAttribute != null ? transactionAttribute : new DefaultTransactionAttribute() { @Override public boolean rollbackOn(Throwable ex) { return true; } - + }; } @@ -327,7 +329,7 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { protected RepeatOperations getStepOperations() { return stepOperations; } - + /** * Public setter for the stepOperations. * @param stepOperations the stepOperations to set @@ -335,7 +337,7 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { public void setStepOperations(RepeatOperations stepOperations) { this.stepOperations = stepOperations; } - + /** * Public setter for the chunkOperations. * @param chunkOperations the chunkOperations to set @@ -399,10 +401,9 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { Assert.notNull(getItemReader(), "ItemReader must be provided"); Assert.notNull(getItemWriter(), "ItemWriter must be provided"); Assert.notNull(transactionManager, "TransactionManager must be provided"); - jobRepositoryValidator.validate(jobRepository); step.setTransactionManager(transactionManager); - if (transactionAttribute!=null) { + if (transactionAttribute != null) { step.setTransactionAttribute(transactionAttribute); } step.setJobRepository(jobRepository); @@ -437,7 +438,12 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { step.registerStepExecutionListener((StepExecutionListener) itemWriter); } - StepExecutionListener[] stepListeners = BatchListenerFactoryHelper.getStepListeners(listeners); + List array = BatchListenerFactoryHelper.getListeners(listeners, + StepExecutionListener.class); + StepExecutionListener[] stepListeners = new StepExecutionListener[array.size()]; + for (int i = 0; i < stepListeners.length; i++) { + stepListeners[i] = array.get(i); + } step.setStepExecutionListeners(stepListeners); if (chunkOperations == null) { @@ -464,8 +470,16 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { step.setStepOperations(stepOperations); - SimpleChunkOrientedTasklet tasklet = new SimpleChunkOrientedTasklet(itemReader, itemProcessor, itemWriter, chunkOperations); - tasklet.setListeners(getListeners()); + SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor(itemProcessor, itemWriter); + chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemProcessListener.class)); + chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemWriteListener.class)); + + SimpleChunkProvider chunkProvider = new SimpleChunkProvider(itemReader, chunkOperations); + @SuppressWarnings("unchecked") + List readListeners = BatchListenerFactoryHelper.getListeners(getListeners(), ItemReadListener.class); + chunkProvider.setListeners(readListeners); + ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + step.setTasklet(tasklet); } @@ -478,7 +492,7 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0), "You must specify either a chunkCompletionPolicy or a commitInterval but not both."); Assert.state(commitInterval >= 0, "The commitInterval must be positive or zero (for default value)."); - + if (chunkCompletionPolicy != null) { return chunkCompletionPolicy; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipWrapper.java similarity index 78% rename from spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java rename to spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipWrapper.java index 996e4cefd..b272f4bce 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipWrapper.java @@ -6,7 +6,7 @@ package org.springframework.batch.core.step.item; * @author Dave Syer * */ -public class ItemWrapper { +public class SkipWrapper { final private Exception exception; @@ -15,12 +15,19 @@ public class ItemWrapper { /** * @param item */ - public ItemWrapper(T item) { + public SkipWrapper(T item) { this(item, null); } + /** + * @param e + */ + public SkipWrapper(Exception e) { + this(null, e); + } - public ItemWrapper(T item, Exception e) { + + public SkipWrapper(T item, Exception e) { this.item = item; this.exception = e; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java deleted file mode 100644 index b6e75abaf..000000000 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java +++ /dev/null @@ -1,87 +0,0 @@ -/* - * 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.core.step.item; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.aop.Advisor; -import org.springframework.aop.framework.Advised; -import org.springframework.batch.item.validator.ValidationException; -import org.springframework.batch.item.validator.Validator; -import org.springframework.transaction.interceptor.TransactionInterceptor; -import org.springframework.util.Assert; - -/** - * Simple validator for internal use only (package private to make it testable). - * Asserts that its argument has no more than the specified number of - * transaction interceptors in its advice chain. - * - * @author Dave Syer - * - */ -class TransactionInterceptorValidator implements Validator { - - protected Log logger = LogFactory.getLog(getClass()); - - private final int maxCount; - - /** - * @param maxCount - */ - public TransactionInterceptorValidator(int maxCount) { - super(); - this.maxCount = maxCount; - } - - /** - * Assert that the object passed in has no more than the maximum number of - * transaction interceptors in its advice chain. - * - * @see org.springframework.batch.item.validator.Validator#validate(java.lang.Object) - */ - public void validate(Object value) throws ValidationException { - Assert.notNull(value, "JobRepository must be provided"); - Assert.state(countTransactionInterceptors(value) <= maxCount, - "JobRepository has more than one transaction interceptor. " - + "Do not declare a separate transaction advice if using the JobRepositoryFactoryBean."); - } - - /** - * @param object an Object, possibly advised - * @return the number of transaction interceptors in the advice chain - */ - private int countTransactionInterceptors(Object object) { - int count = 0; - Object target = object; - while (target instanceof Advised) { - Advised advised = (Advised) target; - Advisor[] interceptors = advised.getAdvisors(); - for (int i = 0; i < interceptors.length; i++) { - if (interceptors[i].getAdvice() instanceof TransactionInterceptor) { - count++; - } - } - try { - target = advised.getTargetSource().getTarget(); - } - catch (Exception e) { - logger.warn("Target could not be obtained from advised instance.", e); - } - } - return count; - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/BatchRetryTemplateTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/BatchRetryTemplateTests.java new file mode 100644 index 000000000..6ecf1b704 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/BatchRetryTemplateTests.java @@ -0,0 +1,200 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.junit.Test; +import org.springframework.batch.retry.ExhaustedRetryException; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryState; +import org.springframework.batch.retry.policy.SimpleRetryPolicy; +import org.springframework.batch.retry.support.DefaultRetryState; + +public class BatchRetryTemplateTests { + + private static class RecoverableException extends Exception { + + public RecoverableException(String message) { + super(message); + } + + } + + private int count = 0; + + private List outputs = new ArrayList(); + + @Test + public void testSuccessfulAttempt() throws Exception { + + BatchRetryTemplate template = new BatchRetryTemplate(); + + String result = template.execute(new RetryCallback() { + public String doWithRetry(RetryContext context) throws Exception { + assertTrue("Wrong context type: " + context.getClass().getSimpleName(), context.getClass().getSimpleName().contains("Batch")); + return "2"; + } + }, Arrays. asList(new DefaultRetryState("1"))); + + assertEquals("2", result); + + } + + @Test + public void testUnSuccessfulAttemptAndRetry() throws Exception { + + BatchRetryTemplate template = new BatchRetryTemplate(); + + RetryCallback retryCallback = new RetryCallback() { + public String[] doWithRetry(RetryContext context) throws Exception { + assertEquals(count, context.getRetryCount()); + if (count++ == 0) { + throw new RecoverableException("Recoverable"); + } + return new String[] { "a", "b" }; + } + }; + + List states = Arrays. asList(new DefaultRetryState("1"), new DefaultRetryState("2")); + try { + template.execute(retryCallback, states); + fail("Expected RecoverableException"); + } + catch (RecoverableException e) { + assertEquals("Recoverable", e.getMessage()); + } + String[] result = template.execute(retryCallback, states); + + assertEquals("[a, b]", Arrays.toString(result)); + + } + + @Test(expected = ExhaustedRetryException.class) + public void testExhaustedRetry() throws Exception { + + BatchRetryTemplate template = new BatchRetryTemplate(); + template.setRetryPolicy(new SimpleRetryPolicy(1)); + + RetryCallback retryCallback = new RetryCallback() { + public String[] doWithRetry(RetryContext context) throws Exception { + if (count++ < 2) { + throw new RecoverableException("Recoverable"); + } + return outputs.toArray(new String[0]); + } + }; + + outputs = Arrays.asList("a", "b"); + try { + template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + fail("Expected RecoverableException"); + } + catch (RecoverableException e) { + assertEquals("Recoverable", e.getMessage()); + } + outputs = Arrays.asList("a", "c"); + template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + + } + + @Test + public void testExhaustedRetryAfterShuffle() throws Exception { + + BatchRetryTemplate template = new BatchRetryTemplate(); + template.setRetryPolicy(new SimpleRetryPolicy(1)); + + RetryCallback retryCallback = new RetryCallback() { + public String[] doWithRetry(RetryContext context) throws Exception { + if (count++ < 1) { + throw new RecoverableException("Recoverable"); + } + return outputs.toArray(new String[0]); + } + }; + + outputs = Arrays.asList("a", "b"); + try { + template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + fail("Expected RecoverableException"); + } + catch (RecoverableException e) { + assertEquals("Recoverable", e.getMessage()); + } + + outputs = Arrays.asList("b", "c"); + try { + template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + fail("Expected ExhaustedRetryException"); + } + catch (ExhaustedRetryException e) { + } + + // "c" is not tarred with same brush as "b" because it was never + // processed on account of the exhausted retry + outputs = Arrays.asList("d", "c"); + String[] result = template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + assertEquals("[d, c]", Arrays.toString(result)); + + // "a" is still marked as a failure from the first chunk + outputs = Arrays.asList("a", "e"); + try { + template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + fail("Expected ExhaustedRetryException"); + } + catch (ExhaustedRetryException e) { + } + + outputs = Arrays.asList("e", "f"); + result = template.execute(retryCallback, BatchRetryTemplate.createState(outputs)); + assertEquals("[e, f]", Arrays.toString(result)); + + } + + @Test + public void testExhaustedRetryWithRecovery() throws Exception { + + BatchRetryTemplate template = new BatchRetryTemplate(); + template.setRetryPolicy(new SimpleRetryPolicy(1)); + + RetryCallback retryCallback = new RetryCallback() { + public String[] doWithRetry(RetryContext context) throws Exception { + if (count++ < 2) { + throw new RecoverableException("Recoverable"); + } + return outputs.toArray(new String[0]); + } + }; + + RecoveryCallback recoveryCallback = new RecoveryCallback() { + public String[] recover(RetryContext context) throws Exception { + List recovered = new ArrayList(); + for (String item : outputs) { + recovered.add("r:"+item); + } + return recovered.toArray(new String[0]); + } + }; + + outputs = Arrays.asList("a", "b"); + try { + template.execute(retryCallback, recoveryCallback, BatchRetryTemplate.createState(outputs)); + fail("Expected RecoverableException"); + } + catch (RecoverableException e) { + assertEquals("Recoverable", e.getMessage()); + } + + outputs = Arrays.asList("b", "c"); + String[] result = template.execute(retryCallback, recoveryCallback, BatchRetryTemplate.createState(outputs)); + assertEquals("[r:b, r:c]", Arrays.toString(result)); + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java new file mode 100644 index 000000000..f72017cb6 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java @@ -0,0 +1,85 @@ +/* + * 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.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.repeat.context.RepeatContextSupport; +import org.springframework.core.AttributeAccessor; + +/** + * @author Dave Syer + * + */ +public class ChunkOrientedTaskletTests { + + private AttributeAccessor context = new RepeatContextSupport(null); + + @Test + public void testHandle() throws Exception { + ChunkOrientedTasklet handler = new ChunkOrientedTasklet(new ChunkProvider() { + public Chunk provide(StepContribution contribution) throws Exception { + contribution.incrementReadCount(); + Chunk chunk = new Chunk(); + chunk.add("foo"); + return chunk; + } + public void postProcess(StepContribution contribution, Chunk chunk) {}; + }, new ChunkProcessor() { + public void process(StepContribution contribution, Chunk chunk) { + contribution.incrementWriteCount(1); + } + }); + StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( + 123L, new JobParameters(), "job")))); + handler.execute(contribution, context); + assertEquals(1, contribution.getReadCount()); + assertEquals(1, contribution.getWriteCount()); + assertEquals(0, context.attributeNames().length); + } + + @Test + public void testFail() throws Exception { + ChunkOrientedTasklet handler = new ChunkOrientedTasklet(new ChunkProvider() { + public Chunk provide(StepContribution contribution) throws Exception { + throw new RuntimeException("Foo!"); + } + public void postProcess(StepContribution contribution, Chunk chunk) {}; + }, new ChunkProcessor() { + public void process(StepContribution contribution, Chunk chunk) { + fail("Not expecting to get this far"); + } + }); + StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( + 123L, new JobParameters(), "job")))); + try { + handler.execute(contribution, context); + fail("Expected RuntimeException"); + } + catch (RuntimeException e) { + assertEquals("Foo!", e.getMessage()); + } + assertEquals(0, contribution.getReadCount()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java deleted file mode 100644 index 5ef6876c3..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java +++ /dev/null @@ -1,468 +0,0 @@ -/* - * 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.core.step.item; - -import static org.junit.Assert.*; -import static org.easymock.EasyMock.*; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.junit.Before; -import org.junit.Test; -import org.springframework.batch.core.SkipListener; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.scope.context.ChunkContext; -import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; -import org.springframework.batch.core.step.skip.SkipPolicy; -import org.springframework.batch.core.step.skip.SkipLimitExceededException; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.NoWorkFoundException; -import org.springframework.batch.item.ParseException; -import org.springframework.batch.item.UnexpectedInputException; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.retry.RetryException; -import org.springframework.batch.retry.policy.NeverRetryPolicy; -import org.springframework.batch.retry.support.RetryTemplate; -import org.springframework.batch.support.Classifier; - -/** - * @author Dave Syer - * - */ -public class FaultTolerantChunkOrientedTaskletTests { - - private Log logger = LogFactory.getLog(getClass()); - - private int count = 0; - - private int limit = 3; - - private int skipLimit = 2; - - private List written = new ArrayList(); - - private List processed = new ArrayList(); - - private FaultTolerantChunkOrientedTasklet tasklet; - - private RepeatTemplate chunkOperations = new RepeatTemplate(); - - private ItemReader itemReader = new ItemReader() { - public Integer read() { - return count++ >= limit ? null : count; - }; - }; - - private ItemWriter itemWriter = new ItemWriter() { - public void write(List items) throws Exception { - written.addAll(items); - } - }; - - private ItemProcessor itemProcessor = new ItemProcessor() { - public String process(Integer item) throws Exception { - return "" + item; - } - }; - - private RetryTemplate retryTemplate = new RetryTemplate(); - - private Classifier rollbackClassifier = new Classifier() { - public Boolean classify(Throwable classifiable) { - return true; - } - }; - - private SkipPolicy readSkipPolicy = new SkipPolicy() { - public boolean shouldSkip(Throwable t, int skipCount) throws SkipLimitExceededException { - if (skipCount < skipLimit) { - return true; - } - throw new SkipLimitExceededException(skipLimit, t); - } - }; - - private SkipPolicy writeSkipPolicy = readSkipPolicy; - - @Before - public void setUp() { - retryTemplate.setRetryPolicy(new NeverRetryPolicy()); - } - - @Test - public void testBasicHandle() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, itemProcessor, itemWriter, - chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - tasklet.execute(contribution, new ChunkContext()); - assertEquals(limit, contribution.getReadCount()); - } - - @Test - public void testSkipOnRead() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(new ItemReader() { - public Integer read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { - throw new RuntimeException("Barf!"); - } - }, itemProcessor, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, - writeSkipPolicy, writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - try { - tasklet.execute(contribution, attributes); - fail("Expected SkipLimitExceededException"); - } - catch (SkipLimitExceededException e) { - // expected - } - assertEquals(0, contribution.getReadCount()); - assertEquals(2, contribution.getReadSkipCount()); - } - - @Test - public void testSkipSingleItemOnWrite() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, itemProcessor, - new ItemWriter() { - public void write(List items) throws Exception { - written.addAll(items); - throw new RuntimeException("Barf!"); - } - }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException"); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - assertTrue(attributes.hasAttribute("SKIPPED_OUTPUTS_KEY")); - tasklet.execute(contribution, attributes); - assertEquals(1, contribution.getReadCount()); - assertEquals(1, contribution.getWriteSkipCount()); - assertEquals(1, written.size()); - } - - @Test - public void testSkipMultipleItemsOnWrite() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, itemProcessor, - new ItemWriter() { - public void write(List items) throws Exception { - logger.debug("Writing items: " + items); - written.addAll(items); - throw new RuntimeException("Barf!"); - } - }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - - // Count to 3: (try + skip + skip) - for (int i = 0; i < 3; i++) { - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException on i=" + i); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - assertTrue(attributes.hasAttribute("SKIPPED_OUTPUTS_KEY")); - } - @SuppressWarnings("unchecked") - Map skips = (Map) attributes.getAttribute("SKIPPED_OUTPUTS_KEY"); - assertEquals(1, skips.size()); - // The last recovery for this chunk... - tasklet.execute(contribution, attributes); - - attributes = new ChunkContext(); - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException"); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - try { - tasklet.execute(contribution, attributes); - fail("Expected SkipLimitExceededException"); - } - catch (SkipLimitExceededException e) { - // expected - } - assertTrue(attributes.hasAttribute("SKIPPED_OUTPUTS_KEY")); - assertEquals(3, contribution.getReadCount()); - assertEquals(0, contribution.getFilterCount()); - assertEquals(2, contribution.getWriteSkipCount()); - assertEquals(5, written.size()); - } - - @Test - public void testSkipSingleItemOnProcess() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, - new ItemProcessor() { - public String process(Integer item) throws Exception { - logger.debug("Processing item: " + item); - processed.add(item); - if (item == 3) { - throw new RuntimeException("Barf!"); - } - return "p" + item; - } - }, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, - writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(3)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - - // try - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException"); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); - - // skip... - tasklet.execute(contribution, attributes); - - assertEquals(3, contribution.getReadCount()); - assertEquals(1, contribution.getProcessSkipCount()); - assertEquals(5, processed.size()); - assertEquals("[p1, p2]", written.toString()); - } - - @Test - public void testSkipOverLimitOnProcess() throws Exception { - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, - new ItemProcessor() { - public String process(Integer item) throws Exception { - logger.debug("Processing item: " + item); - processed.add(item); - throw new RuntimeException("Barf!"); - } - }, itemWriter, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, - writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(2)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - - // Count to 2: (try first + fail) + (skip first + try second + fail) - for (int i = 0; i < 2; i++) { - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException on i=" + i); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); - } - @SuppressWarnings("unchecked") - Map skips = (Map) attributes.getAttribute("SKIPPED_INPUTS_KEY"); - assertEquals(1, skips.size()); - - // The last recovery for this chunk... - tasklet.execute(contribution, attributes); - - attributes = new ChunkContext(); - try { - tasklet.execute(contribution, attributes); - fail("Expected RuntimeException"); - } - catch (Exception e) { - assertEquals("Barf!", e.getMessage()); - } - try { - tasklet.execute(contribution, attributes); - fail("Expected SkipLimitExceededException"); - } - catch (SkipLimitExceededException e) { - // expected - } - assertTrue(attributes.hasAttribute("INPUT_BUFFER_KEY")); - assertEquals(3, contribution.getReadCount()); - assertEquals(2, contribution.getProcessSkipCount()); - // Just before the skip at the end we process once more - assertEquals(3, processed.size()); - } - - /** - * When writer throws an exception that causes rollback, items are - * re-processed in next iteration. - */ - @Test - public void testReprocessAfterWriterRollback() { - final String WRITER_FAILED_MESSAGE = "writer failed"; - final int CHUNK_SIZE = 2; - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, - new ItemProcessor() { - public String process(Integer item) throws Exception { - processed.add(item); - return String.valueOf(item); - } - }, new ItemWriter() { - public void write(List items) throws Exception { - throw new RuntimeException(WRITER_FAILED_MESSAGE); - } - - }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(CHUNK_SIZE)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - - for (int i = 1; i <= 2; i++) { - try { - tasklet.execute(contribution, attributes); - fail(); - } - catch (Exception e) { - assertEquals(WRITER_FAILED_MESSAGE, e.getMessage()); - assertEquals(i * CHUNK_SIZE, processed.size()); - } - } - - } - - /** - * Make sure skip counts are correct when items are skipped on both process - * and write in the same chunk. - */ - @Test - public void testSkipItemOnProcessAndWrite() throws Exception { - final String WRITER_FAILED_MESSAGE = "writer failed"; - final String PROCESSOR_FAILED_MESSAGE = "processor failed"; - final RuntimeException writerException = new RuntimeException(WRITER_FAILED_MESSAGE); - final RuntimeException processorException = new RuntimeException(PROCESSOR_FAILED_MESSAGE); - final int CHUNK_SIZE = 2; - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, - new ItemProcessor() { - public String process(Integer item) throws Exception { - if (item == 1) { - throw processorException; - } - processed.add(item); - return String.valueOf(item); - } - }, new ItemWriter() { - public void write(List items) throws Exception { - throw writerException; - } - - }, chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(CHUNK_SIZE)); - StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - ChunkContext attributes = new ChunkContext(); - - // mock checks skip listener is called as expected - @SuppressWarnings("unchecked") - SkipListener skipListener = createStrictMock(SkipListener.class); - tasklet.registerListener(skipListener); - skipListener.onSkipInProcess(1, processorException); - expectLastCall().once(); - skipListener.onSkipInWrite("2", writerException); - expectLastCall().once(); - replay(skipListener); - - // processor fails first - try { - tasklet.execute(contribution, attributes); - fail(); - } - catch (Exception e) { - assertEquals(PROCESSOR_FAILED_MESSAGE, e.getMessage()); - } - - // we've only rolled back, nothing has been skipped yet - assertEquals(0, contribution.getProcessSkipCount()); - assertEquals(0, contribution.getWriteSkipCount()); - - try { - tasklet.execute(contribution, attributes); - fail(); - } - catch (Exception e) { - assertEquals(WRITER_FAILED_MESSAGE, e.getMessage()); - } - - // processor skipped failed item, writer fails and causes rollback - assertEquals(1, contribution.getProcessSkipCount()); - assertEquals(0, contribution.getWriteSkipCount()); - - tasklet.execute(contribution, attributes); - - // both processor and writer skipped - assertEquals(1, contribution.getProcessSkipCount()); - assertEquals(1, contribution.getWriteSkipCount()); - - verify(skipListener); - } - - @Test - public void testRethrowNonSkippableExceptionOnWriteAsap() throws Exception { - final List chunk = Arrays.asList(new String[] { "1", "2" }); - final Exception ex = new RuntimeException(); - final StepContribution contribution = new StepExecution("foo", null).createStepContribution(); - final Map skipped = new HashMap(); - writeSkipPolicy = new NeverSkipItemSkipPolicy(); - - @SuppressWarnings("unchecked") - ItemWriter itemWriter = createMock(ItemWriter.class); - itemWriter.write(chunk); - expectLastCall().andThrow(ex); - replay(itemWriter); - tasklet = new FaultTolerantChunkOrientedTasklet(itemReader, itemProcessor, itemWriter, - chunkOperations, retryTemplate, rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy); - - try { - tasklet.write(chunk, contribution, skipped); - fail(); - } - catch (Exception e) { - assertSame(ex, e); - } - - try { - tasklet.write(chunk, contribution, skipped); - fail(); - } - catch (Exception e) { - assertTrue(e instanceof RetryException); - assertSame(ex, e.getCause()); - } - - /* - * writer was called only on first failed attempt, exception is rethrown - * immediately when chunk is reprocessed because it is not skippable - */ - verify(itemWriter); - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java new file mode 100644 index 000000000..5781e4da8 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -0,0 +1,54 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.PassthroughItemProcessor; + +public class FaultTolerantChunkProcessorTests { + + private BatchRetryTemplate batchRetryTemplate = new BatchRetryTemplate(); + + private List list = new ArrayList(); + + @Test + public void testWrite() throws Exception { + FaultTolerantChunkProcessor processor = new FaultTolerantChunkProcessor( + new PassthroughItemProcessor(), new ItemWriter() { + public void write(List items) throws Exception { + list.addAll(items); + } + }, batchRetryTemplate); + Chunk inputs = new Chunk(); + inputs.add("1"); + inputs.add("2"); + processor.process(new StepExecution("foo", new JobExecution(0L)).createStepContribution(), inputs); + assertEquals(2, list.size()); + } + + @Test + public void testTransform() throws Exception { + FaultTolerantChunkProcessor processor = new FaultTolerantChunkProcessor( + new ItemProcessor() { + public String process(String item) throws Exception { + return item.equals("1") ? null : item; + } + }, new ItemWriter() { + public void write(List items) throws Exception { + list.addAll(items); + } + }, batchRetryTemplate); + Chunk inputs = new Chunk(); + inputs.add("1"); + inputs.add("2"); + processor.process(new StepExecution("foo", new JobExecution(0L)).createStepContribution(), inputs); + assertEquals(1, list.size()); + } +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java index 66228d487..5dc9492b1 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java @@ -1,12 +1,14 @@ package org.springframework.batch.core.step.item; +import static org.easymock.EasyMock.createStrictMock; +import static org.easymock.EasyMock.expectLastCall; +import static org.easymock.EasyMock.replay; +import static org.easymock.EasyMock.verify; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; -import static org.easymock.EasyMock.*; import java.util.Arrays; import java.util.Collection; -import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -21,16 +23,11 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; -import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.listener.SkipListenerSupport; import org.springframework.batch.core.step.JobRepositorySupport; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; -import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.util.StringUtils; public class FaultTolerantStepFactoryBeanNonBufferingTests { @@ -66,7 +63,7 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests { factory.setSkipLimit(2); factory.setIsReaderTransactionalQueue(true); - JobInstance jobInstance = new JobInstance(1L, new JobParameters(), "skipJob"); + JobInstance jobInstance = new JobInstance(new Long(1), new JobParameters(), "skipJob"); jobExecution = new JobExecution(jobInstance); } @@ -77,12 +74,13 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests { public void testSkip() throws Exception { @SuppressWarnings("unchecked") SkipListener skipListener = createStrictMock(SkipListener.class); + skipListener.onSkipInWrite("3", SkipWriterStub.exception); + expectLastCall().once(); skipListener.onSkipInWrite("4", SkipWriterStub.exception); expectLastCall().once(); replay(skipListener); - + factory.setListeners(new SkipListener[] { skipListener }); - factory.setSkipLimit(1); Step step = (Step) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); @@ -90,199 +88,23 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests { assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); - assertEquals(1, stepExecution.getSkipCount()); - assertEquals(0, stepExecution.getReadSkipCount()); - assertEquals(1, stepExecution.getWriteSkipCount()); - - // only one exception caused rollback, but more than once because it - // has to go back and split the chunk up to isolate the failed item - assertEquals(2, stepExecution.getRollbackCount()); - - assertFalse(writer.written.contains("4")); - - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3,5")); - assertEquals(expectedOutput, writer.written); - - // 5 items + 2 rollbacks re-reading 2 items each time - assertEquals(9, stepExecution.getReadCount()); - - verify(skipListener); - } - - @Test - public void testSkipOverLimit() throws Exception { - SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils - .commaDelimitedListToStringArray("3"))); - processor.rollback = false; - - factory.setItemProcessor(processor); - - factory.setSkipLimit(1); - - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - - step.execute(stepExecution); - - assertEquals(1, stepExecution.getSkipCount()); - - assertFalse(writer.written.contains("4")); - - // failure on "4" tripped the skip limit so only first chunk was written - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2")); - assertEquals(expectedOutput, writer.written); - - } - - /** - * Exception in listener causes failure regardless of skip limit. - * @throws Exception - */ - @Test - public void testSkipListenerFailsOnWrite() throws Exception { - - factory.setSkipLimit(7); // some high limit - factory.setItemReader(reader); - factory.setListeners(new StepListener[] { new SkipListenerSupport() { - @Override - public void onSkipInWrite(String item, Throwable t) { - throw new RuntimeException("oops"); - } - } }); - factory.setSkippableExceptionClasses(Collections.> singleton(Exception.class)); - - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - - step.execute(stepExecution); - assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); - assertEquals("oops", stepExecution.getFailureExceptions().get(0).getCause().getMessage()); - assertEquals(1, stepExecution.getSkipCount()); - assertEquals(0, stepExecution.getReadSkipCount()); - assertEquals(1, stepExecution.getWriteSkipCount()); - - } - - @Test - public void testSkipOnWriteNotDoubleCounted() throws Exception { - - writer = new SkipWriterStub(Arrays.asList(StringUtils.commaDelimitedListToStringArray("4,5"))); - - factory.setSkipLimit(4); - factory.setItemReader(reader); - factory.setItemWriter(writer); - factory.setCommitInterval(3); // includes all expected skips - - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = jobExecution.createStepExecution(step.getName()); - - step.execute(stepExecution); assertEquals(2, stepExecution.getSkipCount()); assertEquals(0, stepExecution.getReadSkipCount()); assertEquals(2, stepExecution.getWriteSkipCount()); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3")); + // only one exception caused rollback, and only once in this case + // because all items in that chunk were skipped immediately + assertEquals(1, stepExecution.getRollbackCount()); + + assertFalse(writer.written.contains("4")); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,5")); assertEquals(expectedOutput, writer.written); - } - - @Test - public void testDefaultSkipPolicy() throws Exception { - factory.setSkippableExceptionClasses(Collections.> singleton(Exception.class)); - factory.setSkipLimit(1); - List items = Arrays.asList(new String[] { "a", "b", "c" }); - ItemReader provider = new ListItemReader(TransactionAwareProxyFactory - .createTransactionalList(items)) { - public String read() { - String item = super.read(); - count++; - if ("b".equals(item)) { - throw new RuntimeException("Read error - planned failure."); - } - return item; - } - }; - factory.setItemReader(provider); - Step step = (Step) factory.getObject(); - - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - step.execute(stepExecution); - - assertEquals(1, stepExecution.getSkipCount()); - // b is processed once and skipped, plus 1, plus c, plus the null at end - assertEquals(4, count); - } - - /** - * Scenario: Exception in processor that shouldn't cause rollback - */ - @Test - public void testProcessorNoRollback() throws Exception { - - factory.setTransactionAttribute(new DefaultTransactionAttribute()); - SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils - .commaDelimitedListToStringArray("1,3"))); - factory.setItemProcessor(processor); - - final Collection NO_FAILURES = Collections.emptyList(); - factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); - - Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - - processor.rollback = false; - step.execute(stepExecution); - assertEquals(2, stepExecution.getSkipCount()); - assertEquals(0, stepExecution.getRollbackCount()); - - } - - /** - * Scenario: Exception in processor that should cause rollback - */ - @Test - public void testProcessorRollback() throws Exception { - SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils - .commaDelimitedListToStringArray("1,3"))); - factory.setItemProcessor(processor); - - final Collection NO_FAILURES = Collections.emptyList(); - factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); - - Step step = (Step) factory.getObject(); - StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - - processor.rollback = true; - step.execute(stepExecution); - assertEquals(2, stepExecution.getSkipCount()); - assertEquals(2, stepExecution.getRollbackCount()); - } - - private static class SkipProcessorStub implements ItemProcessor { - - private final Collection failures; - - private boolean rollback = false; - - public SkipProcessorStub(Collection failures) { - this.failures = failures; - } - - public String process(String item) throws Exception { - if (failures.contains(item)) { - if (rollback) { - throw new SkippableRuntimeException("should cause rollback"); - } - else { - throw new SkippableException("shouldn't cause rollback"); - } - } - return item; - } + // 5 items + 1 rollbacks reading 2 items each time + assertEquals(7, stepExecution.getReadCount()); + verify(skipListener); } /** @@ -299,9 +121,8 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests { private final Collection failures; - @SuppressWarnings("unchecked") public SkipWriterStub() { - this(StringUtils.commaDelimitedListToSet("4")); + this(Arrays.asList("4")); } /** @@ -312,6 +133,7 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests { } public void write(List items) throws Exception { + logger.debug("Writing: " + items); for (String item : items) { if (failures.contains(item)) { logger.debug("Throwing write exception on [" + item + "]"); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java index 77ee384e7..d4550b7ba 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java @@ -15,7 +15,8 @@ */ package org.springframework.batch.core.step.item; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import java.util.ArrayList; import java.util.Arrays; @@ -34,7 +35,6 @@ import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.job.JobSupport; import org.springframework.batch.core.listener.SkipListenerSupport; import org.springframework.batch.core.repository.dao.MapExecutionContextDao; import org.springframework.batch.core.repository.dao.MapJobExecutionDao; @@ -109,11 +109,9 @@ public class FaultTolerantStepFactoryBeanRetryTests { }); factory.setCommitInterval(1); // trivial by default - JobSupport job = new JobSupport("jobName"); - job.setRestartable(true); JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique") .toJobParameters(); - jobExecution = repository.createJobExecution(job.getName(), jobParameters); + jobExecution = repository.createJobExecution("job", jobParameters); jobExecution.setEndTime(new Date()); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index 59c49cff8..73fbc1e24 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -59,6 +59,8 @@ public class FaultTolerantStepFactoryBeanTests { protected int count; + private Collection NO_FAILURES = Collections.emptyList(); + @Before public void setUp() throws Exception { factory.setBeanName("stepName"); @@ -70,7 +72,7 @@ public class FaultTolerantStepFactoryBeanTests { factory.setSkippableExceptionClasses(skippableExceptions); factory.setSkipLimit(2); - JobInstance jobInstance = new JobInstance(1L, new JobParameters(), "skipJob"); + JobInstance jobInstance = new JobInstance(new Long(1), new JobParameters(), "skipJob"); jobExecution = new JobExecution(jobInstance); } @@ -131,29 +133,94 @@ public class FaultTolerantStepFactoryBeanTests { * Check items causing errors are skipped as expected. */ @Test - public void testSkip() throws Exception { + public void testReadSkip() throws Exception { + writer = new SkipWriterStub(NO_FAILURES); + factory.setItemWriter(writer); Step step = (Step) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); - assertEquals(2, stepExecution.getSkipCount()); - assertEquals(1, stepExecution.getReadSkipCount()); - assertEquals(1, stepExecution.getWriteSkipCount()); + System.err.println(writer.written); - // only write exception caused rollback, but more than once because it - // has to go back and split the chunk up to isolate the failed item - assertEquals(2, stepExecution.getRollbackCount()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(1, stepExecution.getReadSkipCount()); + assertEquals(4, stepExecution.getReadCount()); + assertEquals(0, stepExecution.getWriteSkipCount()); + assertEquals(0, stepExecution.getRollbackCount()); // writer did not skip "2" as it never made it to writer, only "4" did assertTrue(reader.processed.contains("4")); - assertFalse(writer.written.contains("4")); + assertFalse(reader.processed.contains("2")); - List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3,5")); + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3,4,5")); + assertEquals(expectedOutput, writer.written); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + + } + + /** + * Check items causing errors are skipped as expected. + */ + @Test + public void testProcessSkip() throws Exception { + + reader = new SkipReaderStub(new String[] { "1", "2", "3", "4", "5" }, NO_FAILURES); + factory.setItemReader(reader); + writer = new SkipWriterStub(NO_FAILURES); + factory.setItemWriter(writer); + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(new String[] { "4" })); + factory.setItemProcessor(processor); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(5, stepExecution.getReadCount()); + assertEquals(1, stepExecution.getProcessSkipCount()); + assertEquals(1, stepExecution.getRollbackCount()); + + // writer skips "4" + assertTrue(reader.processed.contains("4")); + assertFalse(writer.written.contains("4")); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3,5")); + assertEquals(expectedOutput, writer.written); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + + } + + /** + * Check items causing errors are skipped as expected. + */ + @Test + public void testWriteSkip() throws Exception { + + reader = new SkipReaderStub(new String[] { "1", "2", "3", "4", "5" }, NO_FAILURES); + factory.setItemReader(reader); + Step step = (Step) factory.getObject(); + + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(5, stepExecution.getReadCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + assertEquals(2, stepExecution.getRollbackCount()); + + // writer skips "4" + assertTrue(reader.processed.contains("4")); + assertFalse(writer.written.contains("4")); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3,5")); assertEquals(expectedOutput, writer.written); - assertEquals(4, stepExecution.getReadCount()); assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); } @@ -172,7 +239,7 @@ public class FaultTolerantStepFactoryBeanTests { Step step = (Step) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); - + step.execute(stepExecution); assertEquals(1, stepExecution.getSkipCount()); @@ -308,8 +375,8 @@ public class FaultTolerantStepFactoryBeanTests { @Test public void testSkipListenerFailsOnWrite() throws Exception { - reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Arrays - .asList(StringUtils.commaDelimitedListToStringArray("2,3,5"))); + reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), Collections + . emptyList()); factory.setSkipLimit(3); factory.setItemReader(reader); @@ -328,8 +395,8 @@ public class FaultTolerantStepFactoryBeanTests { step.execute(stepExecution); assertEquals(BatchStatus.FAILED, stepExecution.getStatus()); assertEquals("oops", stepExecution.getFailureExceptions().get(0).getCause().getMessage()); - assertEquals(3, stepExecution.getSkipCount()); - assertEquals(2, stepExecution.getReadSkipCount()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); assertEquals(1, stepExecution.getWriteSkipCount()); } @@ -456,8 +523,6 @@ public class FaultTolerantStepFactoryBeanTests { } - // TODO: test with transactional reader (e.g. list with tx proxy) - /** * Scenario: Exception in processor that shouldn't cause rollback */ @@ -468,7 +533,6 @@ public class FaultTolerantStepFactoryBeanTests { .commaDelimitedListToStringArray("1,3"))); factory.setItemProcessor(processor); - final Collection NO_FAILURES = Collections.emptyList(); factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); @@ -491,7 +555,6 @@ public class FaultTolerantStepFactoryBeanTests { .commaDelimitedListToStringArray("1,3"))); factory.setItemProcessor(processor); - final Collection NO_FAILURES = Collections.emptyList(); factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); @@ -512,16 +575,19 @@ public class FaultTolerantStepFactoryBeanTests { return item; } }); - final Collection NO_FAILURES = Collections.emptyList(); factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); Step step = (Step) factory.getObject(); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); - // 1,2,3,4,3,4,3,4 - two re-processing attempts until the item is + + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getRollbackCount()); + + // 1,2,3,4,3,4,3 - two re-processing attempts until the item is // identified and skipped - assertEquals(8, processed.size()); - assertEquals("[1, 2, 3, 4, 3, 4, 3, 4]", processed.toString()); + assertEquals(7, processed.size()); + assertEquals("[1, 2, 3, 4, 3, 4, 3]", processed.toString()); } @@ -603,9 +669,8 @@ public class FaultTolerantStepFactoryBeanTests { private final Collection failures; - @SuppressWarnings("unchecked") public SkipWriterStub() { - this(StringUtils.commaDelimitedListToSet("4")); + this(Arrays.asList("4")); } /** diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTaskletTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTaskletTests.java deleted file mode 100644 index 7c4b36a02..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTaskletTests.java +++ /dev/null @@ -1,173 +0,0 @@ -/* - * 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.core.step.item; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.fail; - -import java.util.List; - -import org.junit.Before; -import org.junit.Test; -import org.springframework.batch.core.JobExecution; -import org.springframework.batch.core.JobInstance; -import org.springframework.batch.core.JobParameters; -import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.StepExecution; -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.item.NoWorkFoundException; -import org.springframework.batch.item.ParseException; -import org.springframework.batch.item.UnexpectedInputException; -import org.springframework.batch.item.support.PassthroughItemProcessor; -import org.springframework.batch.item.validator.ValidationException; -import org.springframework.batch.repeat.context.RepeatContextSupport; -import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.core.AttributeAccessor; - -/** - * @author Dave Syer - * - */ -public class SimpleChunkOrientedTaskletTests { - - private StubItemReader itemReader = new StubItemReader(); - - private StubItemWriter itemWriter = new StubItemWriter(); - - private RepeatTemplate repeatTemplate = new RepeatTemplate(); - - private AttributeAccessor context = new RepeatContextSupport(null); - - @Before - public void setUp() { - repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); - } - - @Test - public void testHandle() throws Exception { - SimpleChunkOrientedTasklet handler = new SimpleChunkOrientedTasklet(itemReader, - new PassthroughItemProcessor(), itemWriter, repeatTemplate); - StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( - 123L, new JobParameters(), "job")))); - handler.execute(contribution, context); - assertEquals(2, itemReader.count); - assertEquals("12", itemWriter.values); - assertEquals(2, contribution.getReadCount()); - assertEquals(2, contribution.getWriteCount()); - assertEquals(0, contribution.getFilterCount()); - } - - @Test - public void testHandleWithItemProcessorFailure() throws Exception { - SimpleChunkOrientedTasklet handler = new SimpleChunkOrientedTasklet(itemReader, - new StubItemProcessor(), itemWriter, repeatTemplate); - StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( - 123L, new JobParameters(), "job")))); - try { - handler.execute(contribution, context); - fail("Expected ValidationException"); - } - catch (ValidationException e) { - // expected - } - assertEquals(2, itemReader.count); - assertEquals(2, contribution.getReadCount()); - assertEquals(0, contribution.getWriteCount()); - assertEquals(0, contribution.getFilterCount()); - assertEquals("", itemWriter.values); - } - - @Test - public void testHandleCompositeItem() throws Exception { - SimpleChunkOrientedTasklet handler = new SimpleChunkOrientedTasklet(itemReader, - new AggregateItemProcessor(), itemWriter, repeatTemplate); - StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance( - 123L, new JobParameters(), "job")))); - handler.execute(contribution, context); - assertEquals(2, itemReader.count); - assertEquals(2, contribution.getReadCount()); - assertEquals(1, contribution.getFilterCount()); - assertEquals(1, contribution.getWriteCount()); - assertEquals("12", itemWriter.values); - } - - - /** - * @author Dave Syer - * - */ - private final class AggregateItemProcessor implements ItemProcessor { - private int count = 0; - - private String value = ""; - - public String process(String item) throws Exception { - value += item; - if (count++ < 1) { - return null; - } - String result = value; - value = ""; - count = 0; - return result; - } - } - - /** - * @author Dave Syer - * - */ - private static class StubItemProcessor implements ItemProcessor { - public String process(String item) throws Exception { - if ("2".equals(item)) { - throw new ValidationException("Planned failure"); - } - return item; - } - } - - /** - * @author Dave Syer - * - */ - private static final class StubItemWriter implements ItemWriter { - private String values = ""; - - public void write(List items) throws Exception { - for (String item : items) { - values += item; - } - } - } - - /** - * @author Dave Syer - * - */ - private final class StubItemReader implements ItemReader { - private int count = 0; - - public String read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException { - if (count++ < 5) - return "" + count; - return null; - } - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java new file mode 100644 index 000000000..d17a6f958 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java @@ -0,0 +1,45 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.PassthroughItemProcessor; + +public class SimpleChunkProcessorTests { + + private SimpleChunkProcessor processor; + + private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution( + new JobInstance(123L, new JobParameters(), "job")))); + + protected List list = new ArrayList(); + + @Before + public void setUp() { + processor = new SimpleChunkProcessor(new PassthroughItemProcessor(), new ItemWriter() { + public void write(List items) throws Exception { + list.addAll(items); + } + }); + } + + @Test + public void testProcess() throws Exception { + Chunk chunk = new Chunk(); + chunk.add("foo"); + chunk.add("bar"); + processor.process(contribution, chunk); + assertEquals(2, list.size()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java new file mode 100644 index 000000000..de2fa3715 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java @@ -0,0 +1,38 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; + +import org.junit.Before; +import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.support.RepeatTemplate; + +public class SimpleChunkProviderTests { + + private SimpleChunkProvider provider; + + private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution( + new JobInstance(123L, new JobParameters(), "job")))); + + @Before + public void setUp() { + provider = new SimpleChunkProvider(new ListItemReader(Arrays.asList("foo", "bar")), + new RepeatTemplate()); + } + + @Test + public void testProvide() throws Exception { + Chunk chunk = provider.provide(contribution); + assertNotNull(chunk); + assertEquals(2, chunk.getItems().size()); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWrapperTests.java similarity index 73% rename from spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java rename to spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWrapperTests.java index 1beabeb32..853a996c6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWrapperTests.java @@ -15,7 +15,8 @@ */ package org.springframework.batch.core.step.item; -import static org.junit.Assert.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import org.junit.Test; @@ -23,36 +24,36 @@ import org.junit.Test; * @author Dave Syer * */ -public class ItemWrapperTests { +public class SkipWrapperTests { private Exception exception = new RuntimeException(); /** - * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object)}. + * Test method for {@link SkipWrapper#SkipWrapper(java.lang.Object)}. */ @Test public void testItemWrapperT() { - ItemWrapper wrapper = new ItemWrapper("foo"); + SkipWrapper wrapper = new SkipWrapper("foo"); assertEquals("foo", wrapper.getItem()); assertEquals(null, wrapper.getException()); } /** - * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#ItemWrapper(java.lang.Object, java.lang.Exception)}. + * Test method for {@link org.springframework.batch.core.step.item.SkipWrapper#SkipWrapper(java.lang.Object, java.lang.Exception)}. */ @Test public void testItemWrapperTException() { - ItemWrapper wrapper = new ItemWrapper("foo",exception); + SkipWrapper wrapper = new SkipWrapper("foo",exception); assertEquals("foo", wrapper.getItem()); assertEquals(exception, wrapper.getException()); } /** - * Test method for {@link org.springframework.batch.core.step.item.ItemWrapper#toString()}. + * Test method for {@link org.springframework.batch.core.step.item.SkipWrapper#toString()}. */ @Test public void testToString() { - ItemWrapper wrapper = new ItemWrapper("foo"); + SkipWrapper wrapper = new SkipWrapper("foo"); assertTrue("foo", wrapper.toString().contains("foo")); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java index f07b9e2f1..06c859e84 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java @@ -3,8 +3,13 @@ */ package org.springframework.batch.core.step.item; -import static org.junit.Assert.*; -import static org.springframework.batch.core.BatchStatus.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.springframework.batch.core.BatchStatus.COMPLETED; +import static org.springframework.batch.core.BatchStatus.FAILED; +import static org.springframework.batch.core.BatchStatus.STOPPED; +import static org.springframework.batch.core.BatchStatus.UNKNOWN; import org.junit.Before; import org.junit.Test; @@ -47,7 +52,7 @@ public class TaskletStepExceptionTests { UpdateCountingJobRepository jobRepository; - static RuntimeException taskletException = new RuntimeException(); + static RuntimeException taskletException = new RuntimeException("Static planned test exception."); static JobInterruptedException interruptedException = new JobInterruptedException(""); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java deleted file mode 100644 index 7ad14afa4..000000000 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * 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.core.step.item; - -import org.springframework.aop.framework.ProxyFactory; -import org.springframework.batch.core.repository.JobRepository; -import org.springframework.transaction.interceptor.TransactionInterceptor; - -import junit.framework.TestCase; - -/** - * @author Dave Syer - * - */ -public class TransactionInterceptorValidatorTests extends TestCase { - - private TransactionInterceptorValidator validator = new TransactionInterceptorValidator(1); - - public void testValidateNull() { - try { - validator.validate(null); - fail("Expected IllegalArgumentException"); - } catch (IllegalArgumentException e) { - String message = e.getMessage(); - assertTrue("Wrong message: "+message, message.indexOf("JobRepository")>=0); - } - } - - public void testValidateWithNoInterceptors() { - validator.validate(new Object()); - } - - public void testValidateAdvisedWithOneInterceptor() { - validator.validate(ProxyFactory.getProxy(JobRepository.class, new TransactionInterceptor())); - } - - public void testValidateAdvisedWithTwoInterceptors() { - Object target = ProxyFactory.getProxy(JobRepository.class, new TransactionInterceptor()); - ProxyFactory factory = new ProxyFactory(); - factory.setTarget(target); - factory.addInterface(JobRepository.class); - factory.addAdvice(new TransactionInterceptor()); - try { - validator.validate(factory.getProxy()); - fail("Expected IllegalStateException"); - } catch (IllegalStateException e) { - String message = e.getMessage(); - assertTrue("Wrong message: "+message, message.indexOf("JobRepository")>=0); - } - } - -} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java index 7002133aa..3035dd6be 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java @@ -15,8 +15,9 @@ */ package org.springframework.batch.core.step.tasklet; -import org.springframework.batch.core.step.item.FaultTolerantChunkOrientedTasklet; -import org.springframework.batch.core.step.item.SimpleChunkOrientedTasklet; +import org.springframework.batch.core.step.item.ChunkOrientedTasklet; +import org.springframework.batch.core.step.item.SimpleChunkProcessor; +import org.springframework.batch.core.step.item.SimpleChunkProvider; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.PassthroughItemProcessor; @@ -31,13 +32,13 @@ import org.springframework.batch.repeat.support.RepeatTemplate; * * @author Dave Syer */ -public class TestingChunkOrientedTasklet extends SimpleChunkOrientedTasklet { +public class TestingChunkOrientedTasklet extends ChunkOrientedTasklet { /** * */ private static final RepeatTemplate repeatTemplate = new RepeatTemplate(); - + static { // It's only for testing, and we don't want any infinite loops... repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(6)); @@ -45,18 +46,20 @@ public class TestingChunkOrientedTasklet extends SimpleChunkOrientedTasklet itemReader, ItemWriter itemWriter) { - super(itemReader, new PassthroughItemProcessor(), itemWriter, repeatTemplate); + this(itemReader, itemWriter, repeatTemplate); } /** * Creates a {@link PassthroughItemProcessor} and uses it to create an - * instance of {@link FaultTolerantChunkOrientedTasklet}. + * instance of {@link Tasklet}. */ - public TestingChunkOrientedTasklet(ItemReader itemReader, ItemWriter itemWriter, RepeatOperations repeatOperations) { - super(itemReader, new PassthroughItemProcessor(), itemWriter, repeatOperations); + public TestingChunkOrientedTasklet(ItemReader itemReader, ItemWriter itemWriter, + RepeatOperations repeatOperations) { + super(new SimpleChunkProvider(itemReader, repeatOperations), new SimpleChunkProcessor( + new PassthroughItemProcessor(), itemWriter)); } }