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

BATCH-1656: fix logic error for skip limit exceeded on no-rollback exception

Browse Source
This commit is contained in:
Dave Syer
2010-11-22 11:03:39 +00:00
parent 9bb970afe2
commit f55bc5e44f
4 changed files with 83 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepFactoryBeanRollbackIntegrationTests.java
View File

@@ -74,7 +74,6 @@ public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
@Autowired
private PlatformTransactionManager transactionManager;
@SuppressWarnings("unchecked")
@Before
public void setUp() throws Exception {
@@ -89,19 +88,11 @@ public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
factory.setJobRepository(repository);
factory.setCommitInterval(3);
factory.setSkipLimit(10);
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
factory.setSkippableExceptionClasses(getExceptionMap(Exception.class));
SimpleJdbcTestUtils.deleteFromTables(new SimpleJdbcTemplate(dataSource), "ERROR_LOG");
}
@Test
public void testUpdatesNoRollback() throws Exception {
@@ -120,12 +111,23 @@ public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
@Test
public void testMultithreadedSkipInWriter() throws Throwable {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(3);
taskExecutor.setMaxPoolSize(6);
taskExecutor.setQueueCapacity(0);
taskExecutor.afterPropertiesSet();
factory.setTaskExecutor(taskExecutor);
@SuppressWarnings("unchecked")
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionMap(Exception.class);
factory.setSkippableExceptionClasses(skippable);
jobExecution = repository.createJobExecution("skipJob", new JobParameters());
for (int i = 0; i < MAX_COUNT; i++) {
if (i%100==0) {
logger.info("Starting step: "+i);
if (i % 100 == 0) {
logger.info("Starting step: " + i);
}
SimpleJdbcTemplate jdbcTemplate = new SimpleJdbcTemplate(dataSource);
@@ -263,7 +265,7 @@ public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
public SkipProcessorStub(DataSource dataSource) {
jdbcTemplate = new SimpleJdbcTemplate(dataSource);
}
/**
* @return the processed
*/
@@ -287,7 +289,7 @@ public class FaultTolerantStepFactoryBeanRollbackIntegrationTests {
public String process(String item) throws Exception {
processed.add(item);
logger.debug("Processed item: "+item);
logger.debug("Processed item: " + item);
jdbcTemplate.update("INSERT INTO ERROR_LOG (MESSAGE, STEP_NAME) VALUES (?, ?)", item, "processed");
return item;
}

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

@@ -29,6 +29,7 @@ import org.springframework.batch.classify.Classifier;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
import org.springframework.batch.core.step.skip.NonSkippableProcessException;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.ItemProcessor;
@@ -269,13 +270,19 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
public O recover(RetryContext context) throws Exception {
Throwable e = context.getLastThrowable();
if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) {
contribution.incrementProcessSkipCount();
iterator.remove(e);
contribution.incrementProcessSkipCount();
logger.debug("Skipping after failed process", e);
return null;
}
else {
throw new RetryException("Non-skippable exception in recoverer while processing", e);
if (rollbackClassifier.classify(e)) {
// Default is to rollback unless the classifier
// allows us to continue
throw new RetryException("Non-skippable exception in recoverer while processing", e);
}
iterator.remove(e);
return null;
}
}
@@ -361,8 +368,8 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
};
batchRetryTemplate.execute(retryCallback, batchRecoveryCallback, BatchRetryTemplate.createState(
getInputKeys(inputs), rollbackClassifier));
batchRetryTemplate.execute(retryCallback, batchRecoveryCallback,
BatchRetryTemplate.createState(getInputKeys(inputs), rollbackClassifier));
}
else {
@@ -456,6 +463,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
try {
return policy.shouldSkip(e, skipCount);
}
catch (SkipLimitExceededException ex) {
throw ex;
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipPolicy.", ex, e);
}

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

@@ -322,6 +322,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
}
@Override
@SuppressWarnings("unchecked")
protected void applyConfiguration(TaskletStep step) {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class,
@@ -520,13 +521,12 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
return skipPolicyWrapper;
}
@SuppressWarnings("unchecked")
private void addNonSkippableExceptionIfMissing(Class... cls) {
List exceptions = new ArrayList<Class<? extends Throwable>>();
for (Class exceptionClass : nonSkippableExceptionClasses) {
private void addNonSkippableExceptionIfMissing(Class<? extends Throwable>... cls) {
List<Class<? extends Throwable>> exceptions = new ArrayList<Class<? extends Throwable>>();
for (Class<? extends Throwable> exceptionClass : nonSkippableExceptionClasses) {
exceptions.add(exceptionClass);
}
for (Class fatal : cls) {
for (Class<? extends Throwable> fatal : cls) {
if (!exceptions.contains(fatal)) {
exceptions.add(fatal);
}
@@ -534,18 +534,17 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
nonSkippableExceptionClasses = exceptions;
}
@SuppressWarnings("unchecked")
private void addNonRetryableExceptionIfMissing(Class... cls) {
List exceptions = new ArrayList<Class<? extends Throwable>>();
for (Class exceptionClass : nonRetryableExceptionClasses) {
private void addNonRetryableExceptionIfMissing(Class<? extends Throwable>... cls) {
List<Class<? extends Throwable>> exceptions = new ArrayList<Class<? extends Throwable>>();
for (Class<? extends Throwable> exceptionClass : nonRetryableExceptionClasses) {
exceptions.add(exceptionClass);
}
for (Class fatal : cls) {
for (Class<? extends Throwable> fatal : cls) {
if (!exceptions.contains(fatal)) {
exceptions.add(fatal);
}
}
nonRetryableExceptionClasses = exceptions;
nonRetryableExceptionClasses = (List<Class<? extends Throwable>>)exceptions;
}
}

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

@@ -4,8 +4,10 @@ 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.HashMap;
import java.util.List;
import java.util.Map;
@@ -196,6 +198,47 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
assertEquals(2, stepExecution.getRollbackCount());
}
@Test
public void testNoRollbackInProcessorWhenSkipExceeded() throws Throwable {
jobExecution = repository.createJobExecution("noRollbackJob", new JobParameters());
factory.setSkipLimit(0);
reader.clear();
reader.setItems("1", "2", "3", "4", "5");
factory.setItemReader(reader);
writer.clear();
factory.setItemWriter(writer);
processor.clear();
factory.setItemProcessor(processor);
@SuppressWarnings("unchecked")
List<Class<? extends Throwable>> exceptions = Arrays.<Class<? extends Throwable>>asList(Exception.class);
factory.setNoRollbackExceptionClasses(exceptions);
@SuppressWarnings("unchecked")
Map<Class<? extends Throwable>, Boolean> skippable = getExceptionMap(Exception.class);
factory.setSkippableExceptionClasses(skippable);
processor.setFailures("2");
Step step = (Step) factory.getObject();
stepExecution = jobExecution.createStepExecution(factory.getName());
repository.add(stepExecution);
step.execute(stepExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals("[1, 3, 4, 5]", writer.getCommitted().toString());
// No rollback on 2 so processor has side effect
assertEquals("[1, 2, 3, 4, 5]", processor.getCommitted().toString());
List<String> processed = new ArrayList<String>(processor.getProcessed());
Collections.sort(processed);
assertEquals("[1, 2, 3, 4, 5]", processed.toString());
assertEquals(0, stepExecution.getSkipCount());
}
@Test
public void testProcessSkipWithNoRollbackForCheckedException() throws Exception {
processor.setFailures("4");