diff --git a/spring-batch-core/.classpath b/spring-batch-core/.classpath index 6be6dfa2c..937d8489b 100644 --- a/spring-batch-core/.classpath +++ b/spring-batch-core/.classpath @@ -4,7 +4,11 @@ - + + + + + diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java index 2af1f52d4..37b0d3c31 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java @@ -15,7 +15,6 @@ */ package org.springframework.batch.core.configuration.xml; -import java.util.Arrays; import java.util.List; import org.springframework.beans.factory.config.BeanDefinition; @@ -129,20 +128,22 @@ public abstract class AbstractStepParser { List txAttrElements = DomUtils.getChildElementsByTagName(stepElement, "transaction-attributes"); if (txAttrElements.size() == 1) { Element txAttrElement = txAttrElements.get(0); - String attributes = DomUtils.getTextValue(txAttrElement); - if (StringUtils.hasLength(attributes)) { - String[] attributesArray = StringUtils.tokenizeToStringArray(attributes, ",\n"); - if (attributesArray.length > 0) { - ManagedList managedList = new ManagedList(); - managedList.setMergeEnabled(Boolean.valueOf(txAttrElement.getAttribute("merge"))); - managedList.addAll(Arrays.asList(attributesArray)); - bd.getPropertyValues().addPropertyValue("transactionAttributeList", managedList); - } + String propagation = txAttrElement.getAttribute("propagation"); + if (StringUtils.hasText(propagation)) { + bd.getPropertyValues().addPropertyValue("propagation", propagation); + } + String isolation = txAttrElement.getAttribute("isolation"); + if (StringUtils.hasText(isolation)) { + bd.getPropertyValues().addPropertyValue("isolation", isolation); + } + String timeout = txAttrElement.getAttribute("timeout"); + if (StringUtils.hasText(timeout)) { + bd.getPropertyValues().addPropertyValue("transactionTimeout", timeout); } } else if (txAttrElements.size() > 1) { parserContext.getReaderContext().error( - "The 'transaction-attribute' element may not appear more than once in a single .", + "The 'transaction-attributes' element may not appear more than once in a single .", stepElement); } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/JobRepositoryParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/JobRepositoryParser.java index 3cb1b0395..4fec5cd11 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/JobRepositoryParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/JobRepositoryParser.java @@ -19,8 +19,8 @@ import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; +import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.util.StringUtils; - import org.w3c.dom.Element; /** @@ -56,7 +56,7 @@ public class JobRepositoryParser extends AbstractSingleBeanDefinitionParser { RuntimeBeanReference tx = new RuntimeBeanReference(transactionManager); builder.addPropertyValue("transactionManager", tx); if (StringUtils.hasText(isolationLevelForCreate)) { - builder.addPropertyValue("isolationLevelForCreate", isolationLevelForCreate); + builder.addPropertyValue("isolationLevelForCreate", DefaultTransactionDefinition.PREFIX_ISOLATION+isolationLevelForCreate); } if (StringUtils.hasText(tablePrefix)) { builder.addPropertyValue("tablePrefix", tablePrefix); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index c60718f57..6a5ba553d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -16,7 +16,6 @@ package org.springframework.batch.core.configuration.xml; -import java.beans.PropertyEditor; import java.util.ArrayList; import java.util.Collection; import java.util.List; @@ -41,10 +40,10 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.core.task.TaskExecutor; import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.interceptor.TransactionAttribute; -import org.springframework.transaction.interceptor.TransactionAttributeEditor; +import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * This {@link FactoryBean} is used by the batch namespace parser to create @@ -65,39 +64,64 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { // Step Attributes // private String name; + private Boolean allowStartIfComplete; + private JobRepository jobRepository; + private Integer startLimit; + private Tasklet tasklet; + private PlatformTransactionManager transactionManager; // // Step Elements // private StepListener[] listeners; - private TransactionAttribute transactionAttribute; + + private int transactionTimeout = DefaultTransactionAttribute.TIMEOUT_DEFAULT; + + private Propagation propagation; + + private Isolation isolation; // // Tasklet Attributes // private Integer cacheCapacity; + private CompletionPolicy chunkCompletionPolicy; + private Integer commitInterval; + private Boolean isReaderTransactionalQueue; + private Integer retryLimit; + private Integer skipLimit; + private TaskExecutor taskExecutor; + private ItemReader itemReader; + private ItemProcessor itemProcessor; + private ItemWriter itemWriter; // // Tasklet Elements // private RetryListener[] retryListeners; + private Collection> skippableExceptionClasses; + private Collection> retryableExceptionClasses; + private Collection> fatalExceptionClasses; + + private Collection> noRollbackExceptionClasses; + private ItemStream[] streams; // @@ -157,8 +181,14 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { if (listeners != null) { fb.setListeners(listeners); } - if (transactionAttribute != null) { - fb.setTransactionAttribute(transactionAttribute); + if (transactionTimeout >= 0) { + fb.setTransactionTimeout(transactionTimeout); + } + if (propagation != null) { + fb.setPropagation(propagation); + } + if (isolation != null) { + fb.setIsolation(isolation); } if (chunkCompletionPolicy != null) { @@ -211,6 +241,9 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { if (fatalExceptionClasses != null) { fb.setFatalExceptionClasses(fatalExceptionClasses); } + if (noRollbackExceptionClasses != null) { + fb.setNoRollbackExceptionClasses(noRollbackExceptionClasses); + } } private void configureTaskletStep(TaskletStep ts) { @@ -240,8 +273,24 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { } ts.setStepExecutionListeners((StepExecutionListener[]) newListeners); } - if (transactionAttribute != null) { - ts.setTransactionAttribute(transactionAttribute); + if (transactionTimeout >= 0 || propagation != null || isolation != null) { + DefaultTransactionAttribute attribute = new DefaultTransactionAttribute(); + attribute.setPropagationBehavior(propagation.value()); + attribute.setIsolationLevel(isolation.value()); + attribute.setTimeout(transactionTimeout); + ts.setTransactionAttribute(new DefaultTransactionAttribute(attribute) { + + /** + * Ignore the default behaviour and rollback on all exceptions that + * bubble up to the tasklet level. The tasklet has to deal with the + * rollback rules internally. + */ + @Override + public boolean rollbackOn(Throwable ex) { + return true; + } + + }); } } @@ -250,6 +299,7 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { new PropertyNamePair(retryListeners, "retry-listeners"), new PropertyNamePair(skippableExceptionClasses, "skippable-exception-classes"), new PropertyNamePair(retryableExceptionClasses, "retryable-exception-classes"), + new PropertyNamePair(noRollbackExceptionClasses, "no-rollback-exception-classes"), new PropertyNamePair(fatalExceptionClasses, "fatal-exception-classes") }; List wrong = new ArrayList(); @@ -268,6 +318,7 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { private static class PropertyNamePair { private Object property; + private String name; public PropertyNamePair(Object property, String name) { @@ -383,24 +434,24 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { } /** - * Set the transaction attribute with a list of all the individual - * attributes. - * - * @param transactionAttributeList + * @param transactionTimeout the transactionTimeout to set */ - public void setTransactionAttributeList(List transactionAttributeList) { - String[] stringArray = transactionAttributeList.toArray(new String[0]); - String attributeString = StringUtils.arrayToCommaDelimitedString(stringArray); - PropertyEditor editor = new TransactionAttributeEditor(); - editor.setAsText(attributeString); - this.setTransactionAttribute((TransactionAttribute) editor.getValue()); + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; } /** - * @param transactionAttribute the {@link TransactionAttribute} to set + * @param isolation the isolation to set */ - public void setTransactionAttribute(TransactionAttribute transactionAttribute) { - this.transactionAttribute = transactionAttribute; + public void setIsolation(Isolation isolation) { + this.isolation = isolation; + } + + /** + * @param propagation the propagation to set + */ + public void setPropagation(Propagation propagation) { + this.propagation = propagation; } // ========================================================= @@ -419,7 +470,7 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { * {@link MapRetryContextCache}.
* * @param cacheCapacity the cache capacity to set (greater than 0 else - * ignored) + * ignored) */ public void setCacheCapacity(int cacheCapacity) { this.cacheCapacity = cacheCapacity; @@ -545,6 +596,14 @@ class StepParserStepFactoryBean implements FactoryBean, BeanNameAware { public void setRetryableExceptionClasses(Collection> retryableExceptionClasses) { this.retryableExceptionClasses = retryableExceptionClasses; } + + /** + * Exception classes that may not cause a rollback if encountered in the right place. + * @param noRollbackExceptionClasses the noRollbackExceptionClasses to set + */ + public void setNoRollbackExceptionClasses(Collection> noRollbackExceptionClasses) { + this.noRollbackExceptionClasses = noRollbackExceptionClasses; + } /** * Public setter for exception classes that should cause immediate failure. diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java index 96ec46510..92a728dc5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java @@ -123,6 +123,8 @@ public class TaskletElementParser { handleExceptionElement(element, parserContext, bd, "fatal-exception-classes", "fatalExceptionClasses"); + handleExceptionElement(element, parserContext, bd, "no-rollback-exception-classes", "noRollbackExceptionClasses"); + handleRetryListenersElement(element, bd, parserContext); handleStreamsElement(element, bd, parserContext); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index c74e1c562..ed4004c01 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java @@ -254,11 +254,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor 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; if (singleton && !inputs.isBusy()) { @@ -280,7 +276,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor inputs, final Chunk outputs, ChunkMonitor chunkMonitor) - throws Exception { + private void scan(final StepContribution contribution, final Chunk inputs, final Chunk outputs, + ChunkMonitor chunkMonitor) throws Exception { if (outputs.isEmpty()) { inputs.setBusy(false); @@ -362,22 +358,17 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor items = Collections.singletonList(outputIterator.next()); try { writeItems(items); + // If successful we are going to return and allow + // the driver to commit... + doAfterWrite(items); + 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); - } } - // If successful we are going to return and allow - // the driver to commit... - doAfterWrite(items); - contribution.incrementWriteCount(1); inputIterator.remove(); outputIterator.remove(); chunkMonitor.incrementOffset(); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java index 240ad6c16..3437067e4 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java @@ -21,8 +21,10 @@ import java.util.Collection; import java.util.HashSet; import java.util.List; +import org.springframework.batch.classify.BinaryExceptionClassifier; import org.springframework.batch.classify.Classifier; import org.springframework.batch.core.JobInterruptedException; +import org.springframework.batch.core.Step; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableReadException; import org.springframework.batch.core.step.skip.SkipLimitExceededException; @@ -69,6 +71,8 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean> skippableExceptionClasses = new HashSet>(); + private Collection> noRollbackExceptionClasses = new HashSet>(); + private Collection> fatalExceptionClasses = new HashSet>(); private Collection> retryableExceptionClasses = new HashSet>(); @@ -187,11 +191,10 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean + * Defaults to all exceptions. * * @param exceptionClasses defaults to Exception */ @@ -211,7 +217,22 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean + * Defaults is empty. + * + * @param noRollbackExceptionClasses the exception classes to set + */ + public void setNoRollbackExceptionClasses(Collection> noRollbackExceptionClasses) { + this.noRollbackExceptionClasses = noRollbackExceptionClasses; + } + + /** + * Exception classes that should cause immediate failure. * * @param fatalExceptionClasses {@link Error} by default */ @@ -219,6 +240,17 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean getRollbackClassifier() { + return new BinaryExceptionClassifier(noRollbackExceptionClasses, false); + } + @Override protected void applyConfiguration(TaskletStep step) { @@ -254,7 +286,8 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean configureChunkProvider() { - SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses, + SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses(), fatalExceptionClasses); FaultTolerantChunkProvider chunkProvider = new FaultTolerantChunkProvider(getItemReader(), getChunkOperations()); @@ -284,15 +317,9 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean configureChunkProcessor() { - SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses, + SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, getSkippableExceptionClasses(), fatalExceptionClasses); - Classifier rollbackClassifier = new Classifier() { - public Boolean classify(Throwable classifiable) { - return getTransactionAttribute().rollbackOn(classifiable); - } - }; - BatchRetryTemplate batchRetryTemplate = configureRetry(); FaultTolerantChunkProcessor chunkProcessor = new FaultTolerantChunkProcessor(getItemProcessor(), @@ -300,7 +327,7 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean> getSkippableExceptionClasses() { + HashSet> set = new HashSet>(skippableExceptionClasses); + set.addAll(noRollbackExceptionClasses); + return set; + } + /** * @return fully configured retry template for item processing phase. */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index baa88b56f..1cb0f3a33 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -43,6 +43,8 @@ import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.core.task.TaskExecutor; import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.TransactionAttribute; import org.springframework.util.Assert; @@ -76,7 +78,11 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { private PlatformTransactionManager transactionManager; - private TransactionAttribute transactionAttribute; + private Propagation propagation = Propagation.REQUIRED; + + private Isolation isolation = Isolation.DEFAULT; + + private int transactionTimeout = DefaultTransactionAttribute.TIMEOUT_DEFAULT; private JobRepository jobRepository; @@ -155,6 +161,30 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { return name; } + /** + * The timeout for an individual transaction in the step. + * + * @param transactionTimeout the transaction timeout to set, defaults to + * infinite + */ + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; + } + + /** + * @param propagation the propagation to set for business transactions + */ + public void setPropagation(Propagation propagation) { + this.propagation = propagation; + } + + /** + * @param isolation the isolation to set for business transactions + */ + public void setIsolation(Isolation isolation) { + this.isolation = isolation; + } + /** * Public setter for the start limit for the step. * @@ -268,27 +298,29 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { } /** - * Public setter for the {@link TransactionAttribute}. - * @param transactionAttribute the {@link TransactionAttribute} to set - */ - public void setTransactionAttribute(TransactionAttribute transactionAttribute) { - this.transactionAttribute = transactionAttribute; - } - - /** - * Protected getter for the {@link TransactionAttribute} for subclasses - * only. + * Getter for the {@link TransactionAttribute} for subclasses only. * @return the transactionAttribute */ protected TransactionAttribute getTransactionAttribute() { - return transactionAttribute != null ? transactionAttribute : new DefaultTransactionAttribute() { + DefaultTransactionAttribute attribute = new DefaultTransactionAttribute(); + attribute.setPropagationBehavior(propagation.value()); + attribute.setIsolationLevel(isolation.value()); + attribute.setTimeout(transactionTimeout); + return new DefaultTransactionAttribute(attribute) { + + /** + * Ignore the default behaviour and rollback on all exceptions that + * bubble up to the tasklet level. The tasklet has to deal with the + * rollback rules internally. + */ @Override public boolean rollbackOn(Throwable ex) { return true; } }; + } /** @@ -407,7 +439,7 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { public void setTaskExecutor(TaskExecutor taskExecutor) { this.taskExecutor = taskExecutor; } - + /** * Mkae the {@link TaskExecutor} available to subclasses * @return the taskExecutor to be used to execute chunks @@ -438,9 +470,7 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { Assert.notNull(transactionManager, "TransactionManager must be provided"); step.setTransactionManager(transactionManager); - if (transactionAttribute != null) { - step.setTransactionAttribute(transactionAttribute); - } + step.setTransactionAttribute(getTransactionAttribute()); step.setJobRepository(jobRepository); step.setStartLimit(startLimit); step.setAllowStartIfComplete(allowStartIfComplete); @@ -530,13 +560,14 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { } return new SimpleCompletionPolicy(commitInterval); } - - private void registerStreams(TaskletStep step, ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter) { + + private void registerStreams(TaskletStep step, ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter) { for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) { if (itemHandler instanceof ItemStream) { - registerStreams(step, new ItemStream[] {(ItemStream) itemHandler}); + registerStreams(step, new ItemStream[] { (ItemStream) itemHandler }); } - } + } } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java index 7048cffe3..3fcac9a99 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/LimitCheckingItemSkipPolicy.java @@ -40,18 +40,19 @@ import org.springframework.batch.item.file.FlatFileParseException; * Furthermore, it is also likely that you only want to skip certain exceptions. * {@link FlatFileParseException} is a good example of an exception you will * likely want to skip, but a {@link FileNotFoundException} should cause - * immediate termination of the {@link Step}. Because it would be impossible - * for a general purpose policy to determine all the types of exceptions that - * should be skipped from those that shouldn't, two lists must be passed in, - * with all of the exceptions that are 'fatal' and 'skippable'. The two lists - * are not enforced to be exclusive, they are prioritized instead - exceptions - * that are fatal will never be skipped, regardless whether the exception can - * also be classified as skippable. + * immediate termination of the {@link Step}. Because it would be impossible for + * a general purpose policy to determine all the types of exceptions that should + * be skipped from those that shouldn't, two lists are passed in, with all + * of the exceptions that are 'fatal' and 'skippable'. The two lists are not + * enforced to be exclusive, they are prioritized instead - exceptions that are + * fatal will never be skipped, regardless whether the exception can also be + * classified as skippable. *

* * @author Ben Hale * @author Lucas Ward * @author Robert Kasanicky + * @author Dave Syer */ public class LimitCheckingItemSkipPolicy implements SkipPolicy { @@ -81,9 +82,24 @@ public class LimitCheckingItemSkipPolicy implements SkipPolicy { */ public LimitCheckingItemSkipPolicy(int skipLimit, Collection> skippableExceptions, Collection> fatalExceptions) { + this(skipLimit, new BinaryExceptionClassifier(skippableExceptions), new BinaryExceptionClassifier( + fatalExceptions)); + } + + /** + * + * @param skipLimit the number of skippable exceptions that are allowed to + * be skipped + * @param skippableExceptionClassifier exception classifier for those that + * can be skipped (non-critical) + * @param fatalExceptionClassifier exception classifier for classes that + * should never be skipped + */ + public LimitCheckingItemSkipPolicy(int skipLimit, Classifier skippableExceptionClassifier, + Classifier fatalExceptionClassifier) { this.skipLimit = skipLimit; - skippableExceptionClassifier = new BinaryExceptionClassifier(skippableExceptions); - fatalExceptionClassifier = new BinaryExceptionClassifier(fatalExceptions); + this.skippableExceptionClassifier = skippableExceptionClassifier; + this.fatalExceptionClassifier = fatalExceptionClassifier; } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 6083be1df..9522b10e5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -260,7 +260,13 @@ public class TaskletStep extends AbstractStep { try { try { - result = tasklet.execute(contribution, chunkContext); + try { + result = tasklet.execute(contribution, chunkContext); + } catch (Exception e) { + if (transactionAttribute.rollbackOn(e)) { + throw e; + } + } chunkListener.afterChunk(); } finally { diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd index 4014f5b7e..9a97f7489 100644 --- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd +++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd @@ -1,11 +1,12 @@ + xmlns:tx="http://www.springframework.org/schema/tx" targetNamespace="http://www.springframework.org/schema/batch" + elementFormDefault="qualified" attributeFormDefault="unqualified" version="2.0"> + - + - + @@ -177,10 +178,12 @@ Defines a stage in job processing backed by a - Step. The id attribute must be specified. The - step requires either a tasklet definition, a + Step. The id attribute must be specified. The + step + requires either a tasklet definition, a tasklet reference, a reference to a step defined - elsewhere, or a reference to a (possibly + elsewhere, or a reference + to a (possibly abstract) parent step. @@ -218,10 +221,10 @@ - + - + @@ -236,7 +239,7 @@ - + @@ -246,13 +249,14 @@ - + The decider is a reference to a JobExecutionDecider that can produce a status to base - the next transition on. + the next + transition on. @@ -268,28 +272,73 @@ - + - - + + - + - + - + - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -351,6 +400,23 @@ + + + + + + + + + + + + + - - - - - - - - - - - @@ -513,9 +564,10 @@ - A reference to a listener, a POJO with a + A reference to a listener, a POJO with a listener-annotated method, or a POJO with - a method referenced by a *-method attribute. + a method + referenced by a *-method attribute. diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java index 2d20f6ec6..15b93e977 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java @@ -34,6 +34,8 @@ import org.springframework.batch.retry.listener.RetryListenerSupport; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.core.task.SyncTaskExecutor; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.annotation.Propagation; /** * @author Dan Garrette @@ -65,7 +67,7 @@ public class StepParserStepFactoryBeanTests { } @Test - public void testTaskletStep_All() throws Exception { + public void testTaskletStepAll() throws Exception { StepParserStepFactoryBean fb = new StepParserStepFactoryBean(); fb.setBeanName("step1"); fb.setAllowStartIfComplete(true); @@ -74,7 +76,9 @@ public class StepParserStepFactoryBeanTests { fb.setTasklet(new DummyTasklet()); fb.setTransactionManager(new ResourcelessTransactionManager()); fb.setListeners(new StepExecutionListenerSupport[] { new StepExecutionListenerSupport() }); - fb.setTransactionAttributeList(new ArrayList()); + fb.setIsolation(Isolation.DEFAULT); + fb.setTransactionTimeout(-1); + fb.setPropagation(Propagation.REQUIRED); Object step = fb.getObject(); assertTrue(step instanceof TaskletStep); Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); @@ -82,7 +86,7 @@ public class StepParserStepFactoryBeanTests { } @Test(expected = IllegalStateException.class) - public void testSimpleStep_All() throws Exception { + public void testSimpleStepAll() throws Exception { StepParserStepFactoryBean fb = new StepParserStepFactoryBean(); fb.setBeanName("step1"); fb.setAllowStartIfComplete(true); @@ -90,7 +94,9 @@ public class StepParserStepFactoryBeanTests { fb.setStartLimit(5); fb.setTransactionManager(new ResourcelessTransactionManager()); fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); - fb.setTransactionAttributeList(new ArrayList()); + fb.setIsolation(Isolation.DEFAULT); + fb.setTransactionTimeout(-1); + fb.setPropagation(Propagation.REQUIRED); fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); fb.setCommitInterval(5); fb.setTaskExecutor(new SyncTaskExecutor()); @@ -113,7 +119,9 @@ public class StepParserStepFactoryBeanTests { fb.setStartLimit(5); fb.setTransactionManager(new ResourcelessTransactionManager()); fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); - fb.setTransactionAttributeList(new ArrayList()); + fb.setIsolation(Isolation.DEFAULT); + fb.setTransactionTimeout(-1); + fb.setPropagation(Propagation.REQUIRED); fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); fb.setCommitInterval(5); fb.setTaskExecutor(new SyncTaskExecutor()); @@ -145,7 +153,9 @@ public class StepParserStepFactoryBeanTests { fb.setStartLimit(5); fb.setTransactionManager(new ResourcelessTransactionManager()); fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); - fb.setTransactionAttributeList(new ArrayList()); + fb.setIsolation(Isolation.DEFAULT); + fb.setTransactionTimeout(-1); + fb.setPropagation(Propagation.REQUIRED); fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); fb.setTaskExecutor(new SyncTaskExecutor()); fb.setItemReader(new DummyItemReader()); @@ -169,7 +179,6 @@ public class StepParserStepFactoryBeanTests { fb.setStartLimit(5); fb.setTransactionManager(new ResourcelessTransactionManager()); fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); - fb.setTransactionAttributeList(new ArrayList()); fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); fb.setTaskExecutor(new SyncTaskExecutor()); fb.setItemReader(new DummyItemReader()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java index 027e18344..083954667 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java @@ -38,8 +38,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.io.ClassPathResource; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.TransactionDefinition; -import org.springframework.transaction.interceptor.RollbackRuleAttribute; -import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; +import org.springframework.transaction.interceptor.DefaultTransactionAttribute; /** * @author Thomas Risberg @@ -53,7 +52,8 @@ public class StepParserTests { "org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml"); Map beans = ctx.getBeansOfType(StepParserStepFactoryBean.class); String factoryName = (String) beans.keySet().toArray()[0]; - StepParserStepFactoryBean factory = (StepParserStepFactoryBean) beans.get(factoryName); + StepParserStepFactoryBean factory = (StepParserStepFactoryBean) beans + .get(factoryName); TaskletStep bean = (TaskletStep) factory.getObject(); assertEquals("wrong start-limit:", 25, bean.getStartLimit()); } @@ -127,7 +127,8 @@ public class StepParserTests { try { new XmlBeanFactory(new ClassPathResource(contextLocation)); fail("Context should not load!"); - } catch (BeanDefinitionParsingException e) { + } + catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().contains("'ref' and 'class'")); } } @@ -162,43 +163,35 @@ public class StepParserTests { "org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml"); // On Inline - No Merge - validateTransactionAttributesInherited("s1", false, ctx); + validateTransactionAttributesInherited("s1", ctx); // On Standalone - No Merge - validateTransactionAttributesInherited("s2", false, ctx); + validateTransactionAttributesInherited("s2", ctx); // On Inline With Tasklet Ref - No Merge - validateTransactionAttributesInherited("s3", false, ctx); + validateTransactionAttributesInherited("s3", ctx); // On Standalone With Tasklet Ref - No Merge - validateTransactionAttributesInherited("s4", false, ctx); + validateTransactionAttributesInherited("s4", ctx); // On Inline - validateTransactionAttributesInherited("s5", true, ctx); + validateTransactionAttributesInherited("s5", ctx); // On Standalone - validateTransactionAttributesInherited("s6", true, ctx); + validateTransactionAttributesInherited("s6", ctx); // On Inline With Tasklet Ref - validateTransactionAttributesInherited("s7", true, ctx); + validateTransactionAttributesInherited("s7", ctx); // On Standalone With Tasklet Ref - validateTransactionAttributesInherited("s8", true, ctx); + validateTransactionAttributesInherited("s8", ctx); } - private void validateTransactionAttributesInherited(String stepName, boolean inherited, ApplicationContext ctx) { - RuleBasedTransactionAttribute txa = getTransactionAttribute(ctx, stepName); + private void validateTransactionAttributesInherited(String stepName, ApplicationContext ctx) { + DefaultTransactionAttribute txa = getTransactionAttribute(ctx, stepName); assertEquals(TransactionDefinition.PROPAGATION_REQUIRED, txa.getPropagationBehavior()); assertEquals(TransactionDefinition.ISOLATION_DEFAULT, txa.getIsolationLevel()); - if (inherited) { - assertEquals(10, txa.getTimeout()); - RollbackRuleAttribute rra = (RollbackRuleAttribute) txa.getRollbackRules().get(0); - assertEquals("org.springframework.dao.DataIntegrityViolationException", rra.getExceptionName()); - } - else { - assertTrue(10 != txa.getTimeout()); - assertTrue(txa.getRollbackRules().isEmpty()); - } + assertEquals(-1, txa.getTimeout()); } @SuppressWarnings("unchecked") @@ -224,7 +217,7 @@ public class StepParserTests { } @SuppressWarnings("unchecked") - private RuleBasedTransactionAttribute getTransactionAttribute(ApplicationContext ctx, String stepName) { + private DefaultTransactionAttribute getTransactionAttribute(ApplicationContext ctx, String stepName) { Map beans = ctx.getBeansOfType(Step.class); assertTrue(beans.containsKey(stepName)); Step step = (Step) ctx.getBean(stepName); @@ -233,7 +226,7 @@ public class StepParserTests { } assertTrue(step instanceof TaskletStep); Object transactionAttribute = ReflectionTestUtils.getField(step, "transactionAttribute"); - RuleBasedTransactionAttribute txa = (RuleBasedTransactionAttribute) transactionAttribute; + DefaultTransactionAttribute txa = (DefaultTransactionAttribute) transactionAttribute; return txa; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java index f02189373..a6dafdec6 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java @@ -37,45 +37,43 @@ import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.test.util.ReflectionTestUtils; -import org.springframework.transaction.TransactionDefinition; -import org.springframework.transaction.interceptor.RollbackRuleAttribute; -import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; - +import org.springframework.transaction.annotation.Isolation; +import org.springframework.transaction.annotation.Propagation; /** * @author Thomas Risberg - * + * */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) public class StepWithFaultTolerantProcessTaskJobParserTests { - + @Autowired private Job job; @Autowired private JobRepository jobRepository; - + @Autowired private TestReader reader; - + @Autowired @Qualifier("listener") private TestListener listener; - + @Autowired private TestRetryListener retryListener; - + @Autowired private TestProcessor processor; - + @Autowired private TestWriter writer; - + @SuppressWarnings("unchecked") @Autowired private StepParserStepFactoryBean factory; - + @Before public void setUp() { MapJobRepositoryFactoryBean.clear(); @@ -92,27 +90,21 @@ public class StepWithFaultTolerantProcessTaskJobParserTests { assertEquals("wrong retry-limit:", 3, rl); Object cc = ReflectionTestUtils.getField(factory, "cacheCapacity"); assertEquals("wrong cache-capacity:", 100, cc); - Object txa = ReflectionTestUtils.getField(factory, "transactionAttribute"); - assertEquals("wrong transaction-attribute:", TransactionDefinition.PROPAGATION_REQUIRED, - ((RuleBasedTransactionAttribute)txa).getPropagationBehavior()); - assertEquals("wrong transaction-attribute:", TransactionDefinition.ISOLATION_DEFAULT, - ((RuleBasedTransactionAttribute)txa).getIsolationLevel()); - assertEquals("wrong transaction-attribute:", 10, - ((RuleBasedTransactionAttribute)txa).getTimeout()); - RollbackRuleAttribute rra = - (RollbackRuleAttribute) ((RuleBasedTransactionAttribute)txa).getRollbackRules().get(0); - assertEquals("wrong transaction-attribute:", - "org.springframework.dao.DataIntegrityViolationException", rra.getExceptionName()); + assertEquals("wrong transaction-attribute:", Propagation.REQUIRED, ReflectionTestUtils.getField(factory, + "propagation")); + assertEquals("wrong transaction-attribute:", Isolation.DEFAULT, ReflectionTestUtils.getField(factory, + "isolation")); + assertEquals("wrong transaction-attribute:", 10, ReflectionTestUtils.getField(factory, "transactionTimeout")); Object txq = ReflectionTestUtils.getField(factory, "isReaderTransactionalQueue"); assertEquals("wrong is-reader-transactional-queue:", true, txq); Object te = ReflectionTestUtils.getField(factory, "taskExecutor"); assertEquals("wrong task-executor:", ConcurrentTaskExecutor.class, te.getClass()); Object listeners = ReflectionTestUtils.getField(factory, "listeners"); - assertEquals("wrong number of listeners:", 2, ((StepListener[])listeners).length); + assertEquals("wrong number of listeners:", 2, ((StepListener[]) listeners).length); Object retryListeners = ReflectionTestUtils.getField(factory, "retryListeners"); - assertEquals("wrong number of retry-listeners:", 2, ((RetryListener[])retryListeners).length); + assertEquals("wrong number of retry-listeners:", 2, ((RetryListener[]) retryListeners).length); Object streams = ReflectionTestUtils.getField(factory, "streams"); - assertEquals("wrong number of streams:", 1, ((ItemStream[])streams).length); + assertEquals("wrong number of streams:", 1, ((ItemStream[]) streams).length); JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); job.execute(jobExecution); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java new file mode 100644 index 000000000..2592c5b4f --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java @@ -0,0 +1,422 @@ +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; + +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.BatchStatus; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; +import org.springframework.transaction.interceptor.RollbackRuleAttribute; +import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.transaction.interceptor.TransactionAttributeEditor; +import org.springframework.util.StringUtils; + +/** + * Tests for {@link FaultTolerantStepFactoryBean}. + */ +public class FaultTolerantStepFactoryBeanRollbackTests { + + protected final Log logger = LogFactory.getLog(getClass()); + + private FaultTolerantStepFactoryBean factory = new FaultTolerantStepFactoryBean(); + + private static Collection NO_FAILURES = Collections.emptyList(); + + private SkipReaderStub reader = new SkipReaderStub(); + + private SkipWriterStub writer = new SkipWriterStub(); + + private JobExecution jobExecution; + + private StepExecution stepExecution; + + private JobRepository repository; + + private static boolean runtimeException = false; + + @Before + public void setUp() throws Exception { + factory.setBeanName("stepName"); + factory.setTransactionManager(new ResourcelessTransactionManager()); + factory.setCommitInterval(2); + factory.setItemReader(reader); + factory.setItemWriter(writer); + factory.setSkipLimit(2); + + MapJobRepositoryFactoryBean.clear(); + MapJobRepositoryFactoryBean repositoryFactory = new MapJobRepositoryFactoryBean(); + repositoryFactory.setTransactionManager(new ResourcelessTransactionManager()); + repositoryFactory.afterPropertiesSet(); + repository = (JobRepository) repositoryFactory.getObject(); + factory.setJobRepository(repository); + + jobExecution = repository.createJobExecution("skipJob", new JobParameters()); + stepExecution = jobExecution.createStepExecution(factory.getName()); + repository.add(stepExecution); + } + + @Test + public void testOverrideWithoutChangingRollbackRules() throws Exception { + TransactionAttributeEditor editor = new TransactionAttributeEditor(); + editor.setAsText("-RuntimeException"); + TransactionAttribute attr = (TransactionAttribute) editor.getValue(); + assertTrue(attr.rollbackOn(new RuntimeException(""))); + assertFalse(attr.rollbackOn(new Exception(""))); + } + + @Test + public void testChangeRollbackRules() throws Exception { + TransactionAttributeEditor editor = new TransactionAttributeEditor(); + editor.setAsText("+RuntimeException"); + TransactionAttribute attr = (TransactionAttribute) editor.getValue(); + assertFalse(attr.rollbackOn(new RuntimeException(""))); + assertFalse(attr.rollbackOn(new Exception(""))); + } + + @SuppressWarnings("unchecked") + @Test + public void testNonDefaultRollbackRules() throws Exception { + TransactionAttributeEditor editor = new TransactionAttributeEditor(); + editor.setAsText("+RuntimeException,+SkippableException"); + RuleBasedTransactionAttribute attr = (RuleBasedTransactionAttribute) editor.getValue(); + attr.getRollbackRules().add(new RollbackRuleAttribute(Exception.class)); + assertTrue(attr.rollbackOn(new Exception(""))); + assertFalse(attr.rollbackOn(new RuntimeException(""))); + assertFalse(attr.rollbackOn(new SkippableException(""))); + } + + /** + * Scenario: Exception in reader that should not cause rollback + */ + @Test + public void testReaderDefaultNoRollbackOnCheckedException() throws Exception { + factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, Arrays.asList("2", "3"))); + + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getRollbackCount()); + } + + /** + * Scenario: Exception in reader that should not cause rollback + */ + @Test + public void testReaderAttributesOverrideSkippableNoRollback() throws Exception { + factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, Arrays.asList("2", "3"))); + + // No skips by default + factory.setSkippableExceptionClasses(new HashSet>()); + // But this one is explicit in the tx-attrs so it should be skipped + factory.setNoRollbackExceptionClasses(getExceptionList(SkippableException.class)); + + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getRollbackCount()); + } + + /** + * Scenario: Exception in processor that should cause rollback because of + * checked exception + */ + @Test + public void testProcessorDefaultRollbackOnCheckedException() throws Exception { + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,3"))); + factory.setItemProcessor(processor); + + factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); + factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); + + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getRollbackCount()); + } + + /** + * Scenario: Exception in processor that should cause rollback + */ + @Test + public void testProcessorDefaultRollbackOnRuntimeException() throws Exception { + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(StringUtils + .commaDelimitedListToStringArray("1,3"))); + factory.setItemProcessor(processor); + + factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); + factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); + + Step step = (Step) factory.getObject(); + + runtimeException = true; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getRollbackCount()); + } + + @Test + public void testProcessSkipWithNoRollbackForCheckedException() throws Exception { + + reader = new SkipReaderStub(new String[] { "1", "2", "3", "4", "5" }, NO_FAILURES); + factory.setItemReader(reader); + factory.setNoRollbackExceptionClasses(getExceptionList(SkippableException.class)); + SkipProcessorStub processor = new SkipProcessorStub(Arrays.asList(new String[] { "4" })); + factory.setItemProcessor(processor); + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(1, stepExecution.getSkipCount()); + assertEquals(0, stepExecution.getReadSkipCount()); + assertEquals(5, stepExecution.getReadCount()); + assertEquals(1, stepExecution.getProcessSkipCount()); + assertEquals(0, stepExecution.getRollbackCount()); + + // skips "4" + assertTrue(reader.processed.contains("4")); + assertFalse(writer.written.contains("4")); + + List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,2,3,5")); + assertEquals(expectedOutput, writer.written); + + } + + /** + * Scenario: Exception in writer that should not cause rollback and scan + */ + @Test + public void testWriterDefaultRollbackOnCheckedException() throws Exception { + factory.setItemWriter(new SkipWriterStub(Arrays.asList("2", "3"))); + + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(4, stepExecution.getRollbackCount()); + } + + /** + * Scenario: Exception in writer that should not cause rollback and scan + */ + @Test + public void testWriterDefaultRollbackOnRuntimeException() throws Exception { + factory.setItemWriter(new SkipWriterStub(Arrays.asList("2", "3"))); + + Step step = (Step) factory.getObject(); + + runtimeException = true; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(4, stepExecution.getRollbackCount()); + + } + + /** + * Scenario: Exception in writer that should not cause rollback and scan + */ + @Test + public void testWriterNoRollbackOnRuntimeException() throws Exception { + factory.setItemWriter(new SkipWriterStub(Arrays.asList("2", "3"))); + factory.setNoRollbackExceptionClasses(getExceptionList(SkippableRuntimeException.class)); + + Step step = (Step) factory.getObject(); + + runtimeException = true; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + // Two multi-item chunks rolled back. When the item was encountered on + // its own it can proceed + assertEquals(2, stepExecution.getRollbackCount()); + + } + + /** + * Scenario: Exception in writer that should not cause rollback and scan + */ + @Test + public void testWriterNoRollbackOnCheckedException() throws Exception { + factory.setItemWriter(new SkipWriterStub(Arrays.asList("2", "3"))); + factory.setNoRollbackExceptionClasses(getExceptionList(SkippableException.class)); + + Step step = (Step) factory.getObject(); + + runtimeException = false; + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(2, stepExecution.getSkipCount()); + // Two multi-item chunks rolled back. When the item was encountered on + // its own it can proceed + assertEquals(2, stepExecution.getRollbackCount()); + + } + + @SuppressWarnings("unchecked") + private Collection> getExceptionList(Class args) { + return Arrays.> asList(args); + } + + private static class SkipProcessorStub implements ItemProcessor { + private final Collection failures; + + public SkipProcessorStub() { + this(NO_FAILURES); + } + + public SkipProcessorStub(Collection failures) { + this.failures = failures; + } + + public String process(String item) throws Exception { + if (failures.contains(item)) { + if (runtimeException) { + throw new SkippableRuntimeException("should cause rollback"); + } + else { + throw new SkippableException("shouldn't cause rollback"); + } + } + return item; + } + } + + /** + * Simple item reader that supports skip functionality. + */ + private static class SkipReaderStub implements ItemReader { + + protected final Log logger = LogFactory.getLog(getClass()); + + private final String[] items; + + private Collection processed = new ArrayList(); + + private int counter = -1; + + private final Collection failures; + + public SkipReaderStub() { + this(new String[] { "1", "2", "3", "4", "5" }, NO_FAILURES); + } + + public SkipReaderStub(String[] items, Collection failures) { + this.items = items; + this.failures = failures; + } + + public String read() throws Exception, UnexpectedInputException, ParseException { + counter++; + if (counter >= items.length) { + logger.debug("Returning null at count=" + counter); + return null; + } + String item = items[counter]; + if (failures.contains(item)) { + logger.debug("Throwing exception for [" + item + "] at count=" + counter); + if (runtimeException) { + throw new SkippableRuntimeException("should cause rollback in reader"); + } + else { + throw new SkippableException("shouldn't cause rollback in reader"); + } + } + processed.add(item); + logger.debug("Returning [" + item + "] at count=" + counter); + return item; + } + + } + + /** + * Simple item writer that supports skip functionality. + */ + private static class SkipWriterStub implements ItemWriter { + + protected final Log logger = LogFactory.getLog(getClass()); + + // simulate transactional output + private List written = TransactionAwareProxyFactory.createTransactionalList(); + + private final Collection failures; + + public SkipWriterStub() { + this(NO_FAILURES); + } + + /** + * @param failures commaDelimitedListToSet + */ + public SkipWriterStub(Collection failures) { + this.failures = failures; + } + + public void write(List items) throws Exception { + for (String item : items) { + if (failures.contains(item)) { + logger.debug("Throwing write exception on [" + item + "]"); + if (runtimeException) { + throw new SkippableRuntimeException("should cause rollback in writer"); + } + else { + throw new SkippableException("shouldn't cause rollback in writer"); + } + } + written.add(item); + } + } + + } + + private static class SkippableException extends Exception { + public SkippableException(String message) { + super(message); + } + } + + private static class SkippableRuntimeException extends RuntimeException { + public SkippableRuntimeException(String message) { + super(message); + } + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index 1ca1884ff..4d55e874c 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -8,7 +8,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; -import java.util.HashSet; import java.util.List; import org.apache.commons.logging.Log; @@ -42,7 +41,6 @@ import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; -import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.util.StringUtils; /** @@ -55,8 +53,8 @@ public class FaultTolerantStepFactoryBeanTests { private FaultTolerantStepFactoryBean factory = new FaultTolerantStepFactoryBean(); @SuppressWarnings("unchecked") - private Collection> skippableExceptions = new HashSet>(Arrays - .> asList(SkippableException.class, SkippableRuntimeException.class)); + private Collection> skippableExceptions = Arrays.> asList( + SkippableException.class, SkippableRuntimeException.class); private SkipReaderStub reader = new SkipReaderStub(); @@ -254,33 +252,6 @@ public class FaultTolerantStepFactoryBeanTests { .getName())); } - /** - * Check that rollback write exception does cause rollback when included on - * transaction attributes as "no rollback for". - */ - @Test - public void testSkipWithoutRethrow() throws Exception { - factory.setTransactionAttribute(new DefaultTransactionAttribute() { - public boolean rollbackOn(Throwable ex) { - return !(ex instanceof SkippableRuntimeException); - }; - }); - Step step = (Step) factory.getObject(); - - step.execute(stepExecution); - - assertEquals(1, stepExecution.getSkipCount()); - assertEquals(1, stepExecution.getReadSkipCount()); - assertEquals(0, stepExecution.getWriteSkipCount()); - - // one rollback for write exception - assertEquals(1, stepExecution.getRollbackCount()); - - assertEquals(4, stepExecution.getReadCount()); - assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step - .getName())); - } - /** * Fatal exception should cause immediate termination regardless of other * skip settings (note the fatal exception is also classified as rollback). @@ -547,51 +518,6 @@ public class FaultTolerantStepFactoryBeanTests { .getName())); } - /** - * 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); - - factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); - factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); - - Step step = (Step) factory.getObject(); - - processor.rollback = false; - step.execute(stepExecution); - assertEquals(2, stepExecution.getSkipCount()); - assertEquals(0, stepExecution.getRollbackCount()); - assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step - .getName())); - } - - /** - * 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); - - factory.setItemReader(new SkipReaderStub(new String[] { "1", "2", "3", "4" }, NO_FAILURES)); - factory.setItemWriter(new SkipWriterStub(NO_FAILURES)); - - Step step = (Step) factory.getObject(); - - processor.rollback = true; - step.execute(stepExecution); - assertEquals(2, stepExecution.getSkipCount()); - assertEquals(2, stepExecution.getRollbackCount()); - assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step - .getName())); - } - @Test public void testReprocessingAfterWriterRollback() throws Exception { factory.setItemProcessor(new ItemProcessor() { @@ -730,7 +656,7 @@ public class FaultTolerantStepFactoryBeanTests { private static class SkipProcessorStub implements ItemProcessor { private final Collection failures; - private boolean rollback = false; + private boolean runtimeException = false; public SkipProcessorStub(Collection failures) { this.failures = failures; @@ -738,7 +664,7 @@ public class FaultTolerantStepFactoryBeanTests { public String process(String item) throws Exception { if (failures.contains(item)) { - if (rollback) { + if (runtimeException) { throw new SkippableRuntimeException("should cause rollback"); } else { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java index 904a5b754..1922cfe6c 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java @@ -35,6 +35,7 @@ import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobInstance; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.job.JobSupport; @@ -45,6 +46,7 @@ import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; +import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.step.JobRepositorySupport; import org.springframework.batch.core.step.StepInterruptionPolicy; import org.springframework.batch.item.ExecutionContext; @@ -54,12 +56,14 @@ import org.springframework.batch.item.ItemStreamException; import org.springframework.batch.item.ItemStreamSupport; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.RepeatStatus; import org.springframework.batch.repeat.policy.DefaultResultCompletionPolicy; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.transaction.TransactionException; +import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.support.DefaultTransactionStatus; public class TaskletStepTests { @@ -785,6 +789,30 @@ public class TaskletStepTests { } + @Test + public void testNoRollbackFor() throws Exception { + + step.setTasklet(new Tasklet() { + public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception { + throw new RuntimeException("Bar"); + } + }); + + JobExecution jobExecutionContext = new JobExecution(jobInstance); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecutionContext); + + DefaultTransactionAttribute transactionAttribute = new DefaultTransactionAttribute() { + @Override + public boolean rollbackOn(Throwable ex) { + return false; + } + }; + step.setTransactionAttribute(transactionAttribute); + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + + } + private static class JobRepositoryStub extends JobRepositorySupport { private int updateCount = 0; diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserBadRetryListenerTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserBadRetryListenerTests-context.xml index 01026d37d..e946d8aa4 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserBadRetryListenerTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserBadRetryListenerTests-context.xml @@ -22,11 +22,11 @@ org.springframework.dao.DataIntegrityViolationException, + + org.springframework.dao.DataIntegrityViolationException + - - PROPAGATION_REQUIRED, ISOLATION_DEFAULT, timeout_10, - -org.springframework.dao.DataIntegrityViolationException - + diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml index cdf07e362..9186b93df 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml @@ -8,51 +8,47 @@ - - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + + - + - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + - + - - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + + - + - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + - + - - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + + - + - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + - - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + + - PROPAGATION_REQUIRED,ISOLATION_DEFAULT + - - timeout_10 - -org.springframework.dao.DataIntegrityViolationException - + - + - - + \ No newline at end of file diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml index c9d4b95c6..11a08afa2 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml @@ -23,11 +23,11 @@ org.springframework.dao.DataIntegrityViolationException + + org.springframework.dao.DataIntegrityViolationException + - - PROPAGATION_REQUIRED,ISOLATION_DEFAULT,timeout_10 - -org.springframework.dao.DataIntegrityViolationException - + diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests-context.xml index bc3752a78..c9ae78b46 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests-context.xml @@ -23,13 +23,11 @@ org.springframework.dao.DataIntegrityViolationException + + org.springframework.dao.DataIntegrityViolationException + - - PROPAGATION_REQUIRED - ISOLATION_DEFAULT - timeout_10 - -org.springframework.dao.DataIntegrityViolationException - + diff --git a/spring-batch-infrastructure/.classpath b/spring-batch-infrastructure/.classpath index 0d0c8ed26..c7f4e6ebb 100644 --- a/spring-batch-infrastructure/.classpath +++ b/spring-batch-infrastructure/.classpath @@ -4,7 +4,11 @@ - + + + + + diff --git a/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml b/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml index 28a54e496..e4a941bb6 100644 --- a/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/skipSampleJob.xml @@ -37,7 +37,6 @@ - java.lang.RuntimeException diff --git a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml index c212cffff..d539cf303 100644 --- a/spring-batch-samples/src/main/resources/jobs/tradeJob.xml +++ b/spring-batch-samples/src/main/resources/jobs/tradeJob.xml @@ -1,9 +1,6 @@ - - + - + - - PROPAGATION_REQUIRED - ISOLATION_READ_COMMITTED - + - + - + @@ -40,15 +31,15 @@ - + - + - + - + - + @@ -68,7 +59,7 @@ + p:dataSource-ref="dataSource"> @@ -77,12 +68,12 @@ + p:dataSource-ref="dataSource" /> - +