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

RESOLVED - issue BATCH-1129: Problems with exception classifications

Browse Source
This commit is contained in:
dsyer
2009-03-20 11:01:54 +00:00
parent de21d44af1
commit 05ed35106c
24 changed files with 900 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java
View File

@@ -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<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
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<String>());
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<Object, Object> fb = new StepParserStepFactoryBean<Object, Object>();
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<String>());
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<String>());
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<String>());
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<String>());
fb.setChunkCompletionPolicy(new DummyCompletionPolicy());
fb.setTaskExecutor(new SyncTaskExecutor());
fb.setItemReader(new DummyItemReader());

43
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java
View File

@@ -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<String, Object> beans = ctx.getBeansOfType(StepParserStepFactoryBean.class);
String factoryName = (String) beans.keySet().toArray()[0];
StepParserStepFactoryBean<Object, Object> factory = (StepParserStepFactoryBean<Object, Object>) beans.get(factoryName);
StepParserStepFactoryBean<Object, Object> factory = (StepParserStepFactoryBean<Object, Object>) 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<String, Object> 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;
}

46
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java
View File

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

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

@@ -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<String, String> factory = new FaultTolerantStepFactoryBean<String, String>();
private static Collection<String> 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<Class<? extends Throwable>>());
// 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<String> 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<Class<? extends Throwable>> getExceptionList(Class<? extends Throwable> args) {
return Arrays.<Class<? extends Throwable>> asList(args);
}
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Collection<String> failures;
public SkipProcessorStub() {
this(NO_FAILURES);
}
public SkipProcessorStub(Collection<String> 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<String> {
protected final Log logger = LogFactory.getLog(getClass());
private final String[] items;
private Collection<String> processed = new ArrayList<String>();
private int counter = -1;
private final Collection<String> failures;
public SkipReaderStub() {
this(new String[] { "1", "2", "3", "4", "5" }, NO_FAILURES);
}
public SkipReaderStub(String[] items, Collection<String> 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<String> {
protected final Log logger = LogFactory.getLog(getClass());
// simulate transactional output
private List<Object> written = TransactionAwareProxyFactory.createTransactionalList();
private final Collection<String> failures;
public SkipWriterStub() {
this(NO_FAILURES);
}
/**
* @param failures commaDelimitedListToSet
*/
public SkipWriterStub(Collection<String> failures) {
this.failures = failures;
}
public void write(List<? extends String> 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);
}
}
}

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

@@ -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<String, String> factory = new FaultTolerantStepFactoryBean<String, String>();
@SuppressWarnings("unchecked")
private Collection<Class<? extends Throwable>> skippableExceptions = new HashSet<Class<? extends Throwable>>(Arrays
.<Class<? extends Throwable>> asList(SkippableException.class, SkippableRuntimeException.class));
private Collection<Class<? extends Throwable>> skippableExceptions = Arrays.<Class<? extends Throwable>> 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<String, String>() {
@@ -730,7 +656,7 @@ public class FaultTolerantStepFactoryBeanTests {
private static class SkipProcessorStub implements ItemProcessor<String, String> {
private final Collection<String> failures;
private boolean rollback = false;
private boolean runtimeException = false;
public SkipProcessorStub(Collection<String> 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 {

28
spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
View File

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