Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS BATCH-919: Draft refactoring introducing ChunkProvider and ChunkProcessor

Browse Source
This commit is contained in:
dsyer
2008-12-10 13:29:35 +00:00
parent c795098827
commit 90c12a7fc3
33 changed files with 1602 additions and 1944 deletions
Download Patch File Download Diff File Expand all files Collapse all files

287
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java
View File

@@ -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 <I> input item type
* @param <O> output item type
*
* @see FaultTolerantChunkOrientedTasklet
* @see NonbufferingFaultTolerantChunkOrientedTasklet
*
* @author Robert Kasanicky
*/
public abstract class AbstractFaultTolerantChunkOrientedTasklet<I, O> extends AbstractItemOrientedTasklet<I, O> {
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<Throwable, Boolean> rollbackClassifier;
public AbstractFaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter,
RetryOperations retryOperations, SkipPolicy readSkipPolicy, SkipPolicy processSkipPolicy,
SkipPolicy writeSkipPolicy, Classifier<Throwable, Boolean> 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<Exception> skippedReads, final Map<I, Exception> skippedInputs,
final Map<O, Exception> skippedOutputs) {
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<I, Exception> 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<O, Exception> 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 <T> List<T> getBufferedList(AttributeAccessor attributes, String key) {
List<T> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new ArrayList<T>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
List<T> casted = (List<T>) 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 <T> Map<T, Exception> getBufferedSkips(AttributeAccessor attributes, String key) {
Map<T, Exception> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new LinkedHashMap<T, Exception>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
Map<T, Exception> casted = (Map<T, Exception>) 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<I> inputs, final List<O> outputs,
final Map<I, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final I item : inputs) {
RetryCallback<O> retryCallback = new RetryCallback<O>() {
public O doWithRetry(RetryContext context) throws Exception {
O output = doProcess(item);
return output;
}
};
RecoveryCallback<O> recoveryCallback = new RecoveryCallback<O>() {
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.<br/>
*
* 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.<br/>
*/
protected void write(final List<O> chunk, final StepContribution contribution, final Map<O, Exception> skipped)
throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
doWrite(chunk);
contribution.incrementWriteCount(chunk.size());
return null;
}
};
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
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));
}
}

117
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java
View File

@@ -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 <I> input item type
* @param <O> output item type
*/
public abstract class AbstractItemOrientedTasklet<I, O> implements Tasklet {
protected final Log logger = LogFactory.getLog(getClass());
protected final ItemReader<? extends I> itemReader;
protected final ItemProcessor<? super I, ? extends O> itemProcessor;
protected final ItemWriter<? super O> itemWriter;
protected final MulticasterBatchListener<I, O> listener = new MulticasterBatchListener<I, O>();
public AbstractItemOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> 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<O> items) throws Exception {
try {
listener.beforeWrite(items);
itemWriter.write(items);
listener.afterWrite(items);
}
catch (Exception e) {
listener.onWriteError(e, items);
throw e;
}
}
}

18
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java
View File

@@ -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<StepExecutionListener> list = new ArrayList<StepExecutionListener>();
public static <T> List<T> getListeners(StepListener[] listeners, Class<? super T> cls) {
List<T> list = new ArrayList<T>();
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;
}
}

254
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchRetryTemplate.java Normal file
View File

@@ -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<RetryState> keys;
public BatchRetryState(Collection<RetryState> keys) {
super(keys);
this.keys = new ArrayList<RetryState>(keys);
}
}
private static class BatchRetryContext extends RetryContextSupport {
private final Collection<RetryContext> contexts;
public BatchRetryContext(RetryContext parent, Collection<RetryContext> 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<RetryContext> contexts = new ArrayList<RetryContext>();
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<RetryContext> 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<RetryContext> contextIterator = batchContext.contexts.iterator();
for (RetryState retryState : batchState.keys) {
RetryContext nextContext = contextIterator.next();
super.close(retryPolicy, nextContext, retryState, succeeded);
}
}
@Override
protected <T> T handleRetryExhausted(RecoveryCallback<T> 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<RetryContext> 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> T execute(RetryCallback<T> retryCallback, Collection<RetryState> states) throws ExhaustedRetryException,
Exception {
RetryState batchState = new BatchRetryState(states);
return delegate.execute(retryCallback, batchState);
}
public <T> T execute(RetryCallback<T> retryCallback, RecoveryCallback<T> recoveryCallback,
Collection<RetryState> states) throws ExhaustedRetryException, Exception {
RetryState batchState = new BatchRetryState(states);
return delegate.execute(retryCallback, recoveryCallback, batchState);
}
public final <T> T execute(RetryCallback<T> retryCallback, RecoveryCallback<T> recoveryCallback,
RetryState retryState) throws Exception, ExhaustedRetryException {
return regular.execute(retryCallback, recoveryCallback, retryState);
}
public final <T> T execute(RetryCallback<T> retryCallback, RecoveryCallback<T> recoveryCallback) throws Exception {
return regular.execute(retryCallback, recoveryCallback);
}
public final <T> T execute(RetryCallback<T> retryCallback, RetryState retryState) throws Exception,
ExhaustedRetryException {
return regular.execute(retryCallback, retryState);
}
public final <T> T execute(RetryCallback<T> retryCallback) throws Exception {
return regular.execute(retryCallback);
}
public static List<RetryState> createState(List<?> keys) {
List<RetryState> states = new ArrayList<RetryState>();
for (Object key : keys) {
states.add(new DefaultRetryState(key));
}
return states;
}
public static List<RetryState> createState(List<?> keys, Classifier<? super Throwable, Boolean> classifier) {
List<RetryState> states = new ArrayList<RetryState>();
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);
}
}

80
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java
View File

@@ -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<W> implements Iterable<W> {
private List<W> items = new ArrayList<W>();
private List<ItemWrapper<W>> skips = new ArrayList<ItemWrapper<W>>();
private List<SkipWrapper<W>> skips = new ArrayList<SkipWrapper<W>>();
private List<Exception> errors = new ArrayList<Exception>();
private Object userData;
private boolean end;
public Chunk() {
this(null,null);
}
public Chunk(List<W> items, List<SkipWrapper<W>> skips) {
super();
if (items!=null) {
this.items = new ArrayList<W>(items);
}
if (skips!=null) {
this.skips = new ArrayList<SkipWrapper<W>>(skips);
}
}
/**
* Add the item to the chunk.
@@ -34,6 +54,8 @@ class Chunk<W> implements Iterable<W> {
*/
public void clear() {
items.clear();
skips.clear();
userData = null;
}
/**
@@ -46,8 +68,25 @@ class Chunk<W> implements Iterable<W> {
/**
* @return a copy of the skips as an unmodifiable list
*/
public List<ItemWrapper<W>> getSkips() {
return Collections.unmodifiableList(new ArrayList<ItemWrapper<W>>(skips));
public List<SkipWrapper<W>> getSkips() {
return Collections.unmodifiableList(skips);
}
/**
* @return a copy of the anonymous errros as an unmodifiable list
*/
public List<Exception> 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<W> implements Iterable<W> {
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<W> implements Iterable<W> {
}
/**
* 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<W> implements Iterable<W> {
}
public void remove(Exception e) {
remove();
skips.add(new SkipWrapper<W>(next, e));
}
public void remove() {
if (next == null) {
if (iterator.hasNext()) {
next = iterator.next();
@@ -117,14 +178,9 @@ class Chunk<W> implements Iterable<W> {
return;
}
}
skips.add(new ItemWrapper<W>(next, e));
iterator.remove();
}
public void remove() {
throw new UnsupportedOperationException("To remove an item you must provide an exception.");
}
}
}

67
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java Normal file
View File

@@ -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 <I> input item type
*/
public class ChunkOrientedTasklet<I> implements Tasklet {
private static final String INPUTS_KEY = "INPUTS";
private final ChunkProcessor<I> chunkProcessor;
private final ChunkProvider<I> chunkProvider;
private boolean buffering = true;
public ChunkOrientedTasklet(ChunkProvider<I> chunkProvider, ChunkProcessor<I> 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<I> inputs = (Chunk<I>) 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());
}
}

9
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProcessor.java Normal file
View File

@@ -0,0 +1,9 @@
package org.springframework.batch.core.step.item;
import org.springframework.batch.core.StepContribution;
public interface ChunkProcessor<I> {
void process(StepContribution contribution, Chunk<I> chunk) throws Exception;
}

11
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProvider.java Normal file
View File

@@ -0,0 +1,11 @@
package org.springframework.batch.core.step.item;
import org.springframework.batch.core.StepContribution;
public interface ChunkProvider<T> {
Chunk<T> provide(StepContribution contribution) throws Exception;
void postProcess(StepContribution contribution, Chunk<T> chunk);
}

162
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java
View File

@@ -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.
*
* <code>ItemProcessor</code> is assumed to be transactional. In case of
* rollback caused by error on write the processing phase will be repeated.
*
* @param <I> input item type
* @param <O> output item type
*
* @author Dave Syer
* @author Robert Kasanicky
*/
public class FaultTolerantChunkOrientedTasklet<I, O> extends AbstractFaultTolerantChunkOrientedTasklet<I, O> {
final static private String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY";
public FaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
Classifier<Throwable, Boolean> 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<I> inputs = getBufferedList(attributes, INPUT_BUFFER_KEY);
final List<O> outputs = new ArrayList<O>();
final List<Exception> 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<I, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
final Map<O, Exception> 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<Exception> 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);
}
}
}
}
}

220
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java Normal file
View File

@@ -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<I, O> extends SimpleChunkProcessor<I, O> {
private SkipPolicy itemProcessSkipPolicy = new LimitCheckingItemSkipPolicy(0);
private SkipPolicy itemWriteSkipPolicy = new LimitCheckingItemSkipPolicy(0);
private final BatchRetryTemplate batchRetryTemplate;
private Classifier<Throwable, Boolean> 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<Throwable, Boolean> rollbackClassifier) {
this.rollbackClassifier = rollbackClassifier;
}
public void setBuffering(boolean buffering) {
this.buffering = buffering;
}
public FaultTolerantChunkProcessor(ItemProcessor<? super I, ? extends O> itemProcessor,
ItemWriter<? super O> itemWriter, BatchRetryTemplate batchRetryTemplate) {
super(itemProcessor, itemWriter);
this.batchRetryTemplate = batchRetryTemplate;
}
@Override
protected Chunk<O> transform(final StepContribution contribution, Chunk<I> inputs) throws Exception {
Chunk<O> outputs = new Chunk<O>();
for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
final I item = iterator.next();
RetryCallback<O> retryCallback = new RetryCallback<O>() {
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<O> recoveryCallback = new RecoveryCallback<O>() {
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<I> inputs, final Chunk<O> outputs)
throws Exception {
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
public Object doWithRetry(RetryContext context) throws Exception {
doWrite(outputs.getItems());
contribution.incrementWriteCount(outputs.size());
return null;
}
};
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
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<I>.ChunkIterator inputIterator = inputs.iterator();
for (Chunk<O>.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<Object> batchRecoveryCallback = new RecoveryCallback<Object>() {
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<I>.ChunkIterator inputIterator = inputs.iterator();
for (Chunk<O>.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<I>.ChunkIterator inputIterator, Chunk<O>.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);
}
}
}

45
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java Normal file
View File

@@ -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<I> extends SimpleChunkProvider<I> {
private SkipPolicy skipPolicy = new LimitCheckingItemSkipPolicy(0);
public FaultTolerantChunkProvider(ItemReader<? extends I> itemReader, RepeatOperations repeatOperations) {
super(itemReader, repeatOperations);
}
public void setSkipPolicy(SkipPolicy SkipPolicy) {
this.skipPolicy = SkipPolicy;
}
@Override
protected I read(StepContribution contribution, Chunk<I> 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);
}
}
}
}
}

76
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
View File

@@ -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<T, S> extends SimpleStepFactoryBean<T,
* zero then all exceptions will be propagated from the chunk and cause the
* step to abort.
*
* Note that if chunks are executed concurrently the number of skips can
* potentially exceed the skip limit and step can still finish successfully.
* This is due to the fact that overall skip count can not be synchronized
* between concurrent chunks while they processing, only on chunk
* boundaries.
*
* @param skipLimit the value to set. Default is 0 (never skip).
*/
public void setSkipLimit(int skipLimit) {
@@ -209,8 +207,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
if (!retryableExceptionClasses.isEmpty()) { // otherwise we
// retry
// all exceptions
// retry all exceptions
simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
}
simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses);
@@ -224,35 +221,31 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
retryPolicy = classifierRetryPolicy;
}
RetryTemplate retryTemplate = new RetryTemplate();
BatchRetryTemplate batchRetryTemplate = new BatchRetryTemplate();
if (backOffPolicy != null) {
retryTemplate.setBackOffPolicy(backOffPolicy);
batchRetryTemplate.setBackOffPolicy(backOffPolicy);
}
retryTemplate.setRetryPolicy(retryPolicy);
Classifier<Throwable, Boolean> rollbackClassifier = new Classifier<Throwable, Boolean>() {
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<Class<? extends Throwable>> exceptions = new ArrayList<Class<? extends Throwable>>(
@@ -262,23 +255,32 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
exceptions.addAll(new ArrayList<Class<? extends Throwable>>(retryableExceptionClasses));
SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions,
new ArrayList<Class<? extends Throwable>>(fatalExceptionClasses));
Classifier<Throwable, Boolean> rollbackClassifier = new Classifier<Throwable, Boolean>() {
public Boolean classify(Throwable classifiable) {
return getTransactionAttribute().rollbackOn(classifiable);
}
};
if (isReaderTransactionalQueue) {
NonbufferingFaultTolerantChunkOrientedTasklet<T, S> tasklet = new NonbufferingFaultTolerantChunkOrientedTasklet<T, S>(
getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate,
rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
tasklet.setListeners(getListeners());
FaultTolerantChunkProvider<T> chunkProvider = new FaultTolerantChunkProvider<T>(getItemReader(),
getChunkOperations());
chunkProvider.setSkipPolicy(readSkipPolicy);
chunkProvider.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemReadListener.class));
chunkProvider.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), SkipListener.class));
step.setTasklet(tasklet);
}
else {
FaultTolerantChunkOrientedTasklet<T, S> tasklet = new FaultTolerantChunkOrientedTasklet<T, S>(
getItemReader(), getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate,
rollbackClassifier, readSkipPolicy, writeSkipPolicy, writeSkipPolicy);
tasklet.setListeners(getListeners());
FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(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<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue);
step.setTasklet(tasklet);
}

174
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java
View File

@@ -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 <I> input item type
* @param <O> output item type
*
* @author Robert Kasanicky
*/
public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends
AbstractFaultTolerantChunkOrientedTasklet<I, O> {
public NonbufferingFaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
Classifier<Throwable, Boolean> 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<I> inputs = new ArrayList<I>();
final List<Exception> 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<I, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
final Map<O, Exception> skippedOutputs = getBufferedSkips(attributes, SKIPPED_OUTPUTS_KEY);
final Set<I> inputsIncludingSkips = new HashSet<I>(inputs.size());
final Set<O> outputsIncludingSkips = new HashSet<O>(inputs.size());
if (!inputs.isEmpty()) {
inputsIncludingSkips.addAll(inputs);
inputs.removeAll(skippedInputs.keySet());
final List<O> outputs = new ArrayList<O>();
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<Exception> skippedReads,
final Map<I, Exception> skippedInputs, final Map<O, Exception> skippedOutputs,
final Set<I> inputsIncludingSkips, final Set<O> 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<Exception> 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;
}
}
}

81
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTasklet.java
View File

@@ -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<I, O> extends AbstractItemOrientedTasklet<I, O> {
private RepeatOperations repeatOperations;
public SimpleChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> 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<I> inputs = new ArrayList<I>();
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<O> outputs = new ArrayList<O>();
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;
}
}

159
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java Normal file
View File

@@ -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<I, O> implements ChunkProcessor<I> {
private final ItemProcessor<? super I, ? extends O> itemProcessor;
private final ItemWriter<? super O> itemWriter;
private final MulticasterBatchListener<I, O> listener = new MulticasterBatchListener<I, O>();
public SimpleChunkProcessor(ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> 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<? extends 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);
}
/**
* @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<O> 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<I> inputs) throws Exception {
// If there is no input we don't have to do anything more
if (inputs.isEmpty()) {
return;
}
Chunk<O> 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<O> skips = (Chunk<O>) inputs.getUserData();
if (skips == null) {
skips = new Chunk<O>();
}
outputs = new Chunk<O>(outputs.getItems(), skips.getSkips());
inputs.setUserData(outputs);
write(contribution, inputs, outputs);
for (SkipWrapper<I> 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<O> 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<I> inputs, Chunk<O> outputs) throws Exception {
doWrite(outputs.getItems());
contribution.incrementWriteCount(outputs.size());
}
protected Chunk<O> transform(StepContribution contribution, Chunk<I> inputs) throws Exception {
Chunk<O> outputs = new Chunk<O>();
for (Chunk<I>.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;
}
}

114
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java Normal file
View File

@@ -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 <I> input item type
*/
public class SimpleChunkProvider<I> implements ChunkProvider<I> {
protected final Log logger = LogFactory.getLog(getClass());
protected final ItemReader<? extends I> itemReader;
private final MulticasterBatchListener<I, ?> listener = new MulticasterBatchListener<I, Object>();
private final RepeatOperations repeatOperations;
public SimpleChunkProvider(ItemReader<? extends I> 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<? extends 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;
}
}
public Chunk<I> provide(final StepContribution contribution) throws Exception {
final Chunk<I> inputs = new Chunk<I>();
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<I> 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<I> chunk) throws Exception {
return doRead();
}
}

52
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -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<T,S> implements FactoryBean, BeanNameAware {
public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
private static final int DEFAULT_COMMIT_INTERVAL = 1;
@@ -69,15 +71,13 @@ public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
private ItemWriter<? super S> 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<T,S> implements FactoryBean, BeanNameAware {
private ItemProcessor<? super T, ? extends S> itemProcessor = new ItemProcessor<T, S>() {
@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<T,S> 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<T,S> 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<T,S> 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<T,S> 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<T,S> implements FactoryBean, BeanNameAware {
step.registerStepExecutionListener((StepExecutionListener) itemWriter);
}
StepExecutionListener[] stepListeners = BatchListenerFactoryHelper.getStepListeners(listeners);
List<StepExecutionListener> 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<T,S> implements FactoryBean, BeanNameAware {
step.setStepOperations(stepOperations);
SimpleChunkOrientedTasklet<T,S> tasklet = new SimpleChunkOrientedTasklet<T,S>(itemReader, itemProcessor, itemWriter, chunkOperations);
tasklet.setListeners(getListeners());
SimpleChunkProcessor<T, S> chunkProcessor = new SimpleChunkProcessor<T, S>(itemProcessor, itemWriter);
chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemProcessListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.getListeners(getListeners(), ItemWriteListener.class));
SimpleChunkProvider<T> chunkProvider = new SimpleChunkProvider<T>(itemReader, chunkOperations);
@SuppressWarnings("unchecked")
List<ItemReadListener> readListeners = BatchListenerFactoryHelper.<ItemReadListener>getListeners(getListeners(), ItemReadListener.class);
chunkProvider.setListeners(readListeners);
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
step.setTasklet(tasklet);
}
@@ -478,7 +492,7 @@ public class SimpleStepFactoryBean<T,S> 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;
}

13
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemWrapper.java → spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipWrapper.java
View File

@@ -6,7 +6,7 @@ package org.springframework.batch.core.step.item;
* @author Dave Syer
*
*/
public class ItemWrapper<T> {
public class SkipWrapper<T> {
final private Exception exception;
@@ -15,12 +15,19 @@ public class ItemWrapper<T> {
/**
* @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;
}

87
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/TransactionInterceptorValidator.java
View File

@@ -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;
}
}

200
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/BatchRetryTemplateTests.java Normal file
View File

@@ -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<String> outputs = new ArrayList<String>();
@Test
public void testSuccessfulAttempt() throws Exception {
BatchRetryTemplate template = new BatchRetryTemplate();
String result = template.execute(new RetryCallback<String>() {
public String doWithRetry(RetryContext context) throws Exception {
assertTrue("Wrong context type: " + context.getClass().getSimpleName(), context.getClass().getSimpleName().contains("Batch"));
return "2";
}
}, Arrays.<RetryState> asList(new DefaultRetryState("1")));
assertEquals("2", result);
}
@Test
public void testUnSuccessfulAttemptAndRetry() throws Exception {
BatchRetryTemplate template = new BatchRetryTemplate();
RetryCallback<String[]> retryCallback = new RetryCallback<String[]>() {
public String[] doWithRetry(RetryContext context) throws Exception {
assertEquals(count, context.getRetryCount());
if (count++ == 0) {
throw new RecoverableException("Recoverable");
}
return new String[] { "a", "b" };
}
};
List<RetryState> states = Arrays.<RetryState> 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<String[]> retryCallback = new RetryCallback<String[]>() {
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<String[]> retryCallback = new RetryCallback<String[]>() {
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<String[]> retryCallback = new RetryCallback<String[]>() {
public String[] doWithRetry(RetryContext context) throws Exception {
if (count++ < 2) {
throw new RecoverableException("Recoverable");
}
return outputs.toArray(new String[0]);
}
};
RecoveryCallback<String[]> recoveryCallback = new RecoveryCallback<String[]>() {
public String[] recover(RetryContext context) throws Exception {
List<String> recovered = new ArrayList<String>();
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));
}
}

85
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java Normal file
View File

@@ -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<String> handler = new ChunkOrientedTasklet<String>(new ChunkProvider<String>() {
public Chunk<String> provide(StepContribution contribution) throws Exception {
contribution.incrementReadCount();
Chunk<String> chunk = new Chunk<String>();
chunk.add("foo");
return chunk;
}
public void postProcess(StepContribution contribution, Chunk<String> chunk) {};
}, new ChunkProcessor<String>() {
public void process(StepContribution contribution, Chunk<String> 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<String> handler = new ChunkOrientedTasklet<String>(new ChunkProvider<String>() {
public Chunk<String> provide(StepContribution contribution) throws Exception {
throw new RuntimeException("Foo!");
}
public void postProcess(StepContribution contribution, Chunk<String> chunk) {};
}, new ChunkProcessor<String>() {
public void process(StepContribution contribution, Chunk<String> 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());
}
}

468
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTaskletTests.java
View File

@@ -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<String> written = new ArrayList<String>();
private List<Integer> processed = new ArrayList<Integer>();
private FaultTolerantChunkOrientedTasklet<Integer, String> tasklet;
private RepeatTemplate chunkOperations = new RepeatTemplate();
private ItemReader<Integer> itemReader = new ItemReader<Integer>() {
public Integer read() {
return count++ >= limit ? null : count;
};
};
private ItemWriter<String> itemWriter = new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
written.addAll(items);
}
};
private ItemProcessor<Integer, String> itemProcessor = new ItemProcessor<Integer, String>() {
public String process(Integer item) throws Exception {
return "" + item;
}
};
private RetryTemplate retryTemplate = new RetryTemplate();
private Classifier<Throwable, Boolean> rollbackClassifier = new Classifier<Throwable, Boolean>() {
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<Integer, String>(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<Integer, String>(new ItemReader<Integer>() {
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<Integer, String>(itemReader, itemProcessor,
new ItemWriter<String>() {
public void write(List<? extends String> 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<Integer, String>(itemReader, itemProcessor,
new ItemWriter<String>() {
public void write(List<? extends String> 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<String, Exception> skips = (Map<String, Exception>) 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<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
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<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
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<Integer, Exception> skips = (Map<Integer, Exception>) 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<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
public String process(Integer item) throws Exception {
processed.add(item);
return String.valueOf(item);
}
}, new ItemWriter<String>() {
public void write(List<? extends String> 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<Integer, String>(itemReader,
new ItemProcessor<Integer, String>() {
public String process(Integer item) throws Exception {
if (item == 1) {
throw processorException;
}
processed.add(item);
return String.valueOf(item);
}
}, new ItemWriter<String>() {
public void write(List<? extends String> 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<Integer, String> 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<String> chunk = Arrays.asList(new String[] { "1", "2" });
final Exception ex = new RuntimeException();
final StepContribution contribution = new StepExecution("foo", null).createStepContribution();
final Map<String, Exception> skipped = new HashMap<String, Exception>();
writeSkipPolicy = new NeverSkipItemSkipPolicy();
@SuppressWarnings("unchecked")
ItemWriter<String> itemWriter = createMock(ItemWriter.class);
itemWriter.write(chunk);
expectLastCall().andThrow(ex);
replay(itemWriter);
tasklet = new FaultTolerantChunkOrientedTasklet<Integer, String>(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);
}
}

54
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java Normal file
View File

@@ -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<String> list = new ArrayList<String>();
@Test
public void testWrite() throws Exception {
FaultTolerantChunkProcessor<String, String> processor = new FaultTolerantChunkProcessor<String, String>(
new PassthroughItemProcessor<String>(), new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
list.addAll(items);
}
}, batchRetryTemplate);
Chunk<String> inputs = new Chunk<String>();
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<String, String> processor = new FaultTolerantChunkProcessor<String, String>(
new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
return item.equals("1") ? null : item;
}
}, new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
list.addAll(items);
}
}, batchRetryTemplate);
Chunk<String> inputs = new Chunk<String>();
inputs.add("1");
inputs.add("2");
processor.process(new StepExecution("foo", new JobExecution(0L)).createStepContribution(), inputs);
assertEquals(1, list.size());
}
}

218
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanNonBufferingTests.java
View File

@@ -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<Integer, String> 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<String> 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<String> 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<String, String>() {
@Override
public void onSkipInWrite(String item, Throwable t) {
throw new RuntimeException("oops");
}
} });
factory.setSkippableExceptionClasses(Collections.<Class<? extends Throwable>> 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<String> 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<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,5"));
assertEquals(expectedOutput, writer.written);
}
@Test
public void testDefaultSkipPolicy() throws Exception {
factory.setSkippableExceptionClasses(Collections.<Class<? extends Throwable>> singleton(Exception.class));
factory.setSkipLimit(1);
List<String> items = Arrays.asList(new String[] { "a", "b", "c" });
ItemReader<String> provider = new ListItemReader<String>(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<String> 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<String> 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<String, String> {
private final Collection<String> failures;
private boolean rollback = false;
public SkipProcessorStub(Collection<String> 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<String> failures;
@SuppressWarnings("unchecked")
public SkipWriterStub() {
this(StringUtils.commaDelimitedListToSet("4"));
this(Arrays.asList("4"));
}
/**
@@ -312,6 +133,7 @@ public class FaultTolerantStepFactoryBeanNonBufferingTests {
}
public void write(List<? extends String> items) throws Exception {
logger.debug("Writing: " + items);
for (String item : items) {
if (failures.contains(item)) {
logger.debug("Throwing write exception on [" + item + "]");

8
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java
View File

@@ -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());
}

117
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java
View File

@@ -59,6 +59,8 @@ public class FaultTolerantStepFactoryBeanTests {
protected int count;
private Collection<String> 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<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,3,5"));
List<String> 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<String> 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<String> 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
.<String> 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<String> 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<String> 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<String> 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<String> failures;
@SuppressWarnings("unchecked")
public SkipWriterStub() {
this(StringUtils.commaDelimitedListToSet("4"));
this(Arrays.asList("4"));
}
/**

173
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkOrientedTaskletTests.java
View File

@@ -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<String, String> handler = new SimpleChunkOrientedTasklet<String, String>(itemReader,
new PassthroughItemProcessor<String>(), 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<String, String> handler = new SimpleChunkOrientedTasklet<String, String>(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<String, String> handler = new SimpleChunkOrientedTasklet<String, String>(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<String, String> {
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<String, String> {
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<String> {
private String values = "";
public void write(List<? extends String> items) throws Exception {
for (String item : items) {
values += item;
}
}
}
/**
* @author Dave Syer
*
*/
private final class StubItemReader implements ItemReader<String> {
private int count = 0;
public String read() throws Exception, UnexpectedInputException, NoWorkFoundException, ParseException {
if (count++ < 5)
return "" + count;
return null;
}
}
}

45
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java Normal file
View File

@@ -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<String, String> processor;
private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(
new JobInstance(123L, new JobParameters(), "job"))));
protected List<String> list = new ArrayList<String>();
@Before
public void setUp() {
processor = new SimpleChunkProcessor<String,String>(new PassthroughItemProcessor<String>(), new ItemWriter<String>() {
public void write(List<? extends String> items) throws Exception {
list.addAll(items);
}
});
}
@Test
public void testProcess() throws Exception {
Chunk<String> chunk = new Chunk<String>();
chunk.add("foo");
chunk.add("bar");
processor.process(contribution, chunk);
assertEquals(2, list.size());
}
}

38
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java Normal file
View File

@@ -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<String> provider;
private StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(
new JobInstance(123L, new JobParameters(), "job"))));
@Before
public void setUp() {
provider = new SimpleChunkProvider<String>(new ListItemReader<String>(Arrays.asList("foo", "bar")),
new RepeatTemplate());
}
@Test
public void testProvide() throws Exception {
Chunk<String> chunk = provider.provide(contribution);
assertNotNull(chunk);
assertEquals(2, chunk.getItems().size());
}
}

17
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemWrapperTests.java → spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWrapperTests.java
View File

@@ -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<String> wrapper = new ItemWrapper<String>("foo");
SkipWrapper<String> wrapper = new SkipWrapper<String>("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<String> wrapper = new ItemWrapper<String>("foo",exception);
SkipWrapper<String> wrapper = new SkipWrapper<String>("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<String> wrapper = new ItemWrapper<String>("foo");
SkipWrapper<String> wrapper = new SkipWrapper<String>("foo");
assertTrue("foo", wrapper.toString().contains("foo"));
}

11
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java
View File

@@ -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("");

65
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TransactionInterceptorValidatorTests.java
View File

@@ -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);
}
}
}

21
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TestingChunkOrientedTasklet.java
View File

@@ -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<T> extends SimpleChunkOrientedTasklet<T, T> {
public class TestingChunkOrientedTasklet<T> extends ChunkOrientedTasklet<T> {
/**
*
*/
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<T> extends SimpleChunkOrientedTasklet<T
/**
* Creates a {@link PassthroughItemProcessor} and uses it to create an
* instance of {@link FaultTolerantChunkOrientedTasklet}.
* instance of {@link Tasklet}.
*/
public TestingChunkOrientedTasklet(ItemReader<T> itemReader, ItemWriter<T> itemWriter) {
super(itemReader, new PassthroughItemProcessor<T>(), 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<T> itemReader, ItemWriter<T> itemWriter, RepeatOperations repeatOperations) {
super(itemReader, new PassthroughItemProcessor<T>(), itemWriter, repeatOperations);
public TestingChunkOrientedTasklet(ItemReader<T> itemReader, ItemWriter<T> itemWriter,
RepeatOperations repeatOperations) {
super(new SimpleChunkProvider<T>(itemReader, repeatOperations), new SimpleChunkProcessor<T, T>(
new PassthroughItemProcessor<T>(), itemWriter));
}
}