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

RESOLVED: BATCH-1327 Check for skippable before scanning for failed item

Browse Source
This commit is contained in:
dsyer
2009-07-10 08:07:22 +00:00
parent d9791d60c5
commit 155b51cec0
10 changed files with 187 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
View File

@@ -17,6 +17,8 @@
package org.springframework.batch.core.configuration.xml;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import org.springframework.batch.classify.BinaryExceptionClassifier;
import org.springframework.batch.core.Step;
@@ -283,7 +285,8 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
ts.setStepExecutionListeners((StepExecutionListener[]) newListeners);
}
if (transactionTimeout != null || propagation != null || isolation != null || noRollbackExceptionClasses!=null) {
if (transactionTimeout != null || propagation != null || isolation != null
|| noRollbackExceptionClasses != null) {
DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
if (propagation != null) {
attribute.setPropagationBehavior(propagation.value());
@@ -294,7 +297,9 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
if (transactionTimeout != null) {
attribute.setTimeout(transactionTimeout);
}
final BinaryExceptionClassifier classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false);
Collection<Class<? extends Throwable>> exceptions = noRollbackExceptionClasses == null ? new HashSet<Class<? extends Throwable>>()
: noRollbackExceptionClasses;
final BinaryExceptionClassifier classifier = new BinaryExceptionClassifier(exceptions, false);
ts.setTransactionAttribute(new DefaultTransactionAttribute(attribute) {
@Override
public boolean rollbackOn(Throwable ex) {

92
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java.svntmp
View File

@@ -1,92 +0,0 @@
/*
* Copyright 2006-2009 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
/**
* A {@link Tasklet} implementing variations on read-process-write item
* handling.
*
* @author Dave Syer
*
* @param <I> input item type
*/
public class ChunkOrientedTasklet<I> implements Tasklet {
private static final String INPUTS_KEY = "INPUTS";
private final ChunkProcessor<I> chunkProcessor;
private final ChunkProvider<I> chunkProvider;
private boolean buffering = true;
public ChunkOrientedTasklet(ChunkProvider<I> chunkProvider, ChunkProcessor<I> chunkProcessor) {
this.chunkProvider = chunkProvider;
this.chunkProcessor = chunkProcessor;
}
/**
* Flag to indicate that items should be buffered once read. Defaults to
* true, which is appropriate for forward-only, non-transactional item
* readers. Main (or only) use case for setting this flag to true is a
* transactional JMS item reader.
*
* @param buffering
*/
public void setBuffering(boolean buffering) {
this.buffering = buffering;
}
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
@SuppressWarnings("unchecked")
Chunk<I> inputs = (Chunk<I>) chunkContext.getAttribute(INPUTS_KEY);
if (inputs == null) {
inputs = chunkProvider.provide(contribution);
if (buffering) {
chunkContext.setAttribute(INPUTS_KEY, inputs);
}
}
chunkProcessor.process(contribution, inputs);
chunkProvider.postProcess(contribution, inputs);
// Allow a message coming back from the processor to say that we
// are not done yet
if (inputs.isBusy()) {
// TODO: update ExecutionContext with an offset if the
// ItemReader was stateful
return RepeatStatus.CONTINUABLE;
}
chunkContext.removeAttribute(INPUTS_KEY);
chunkContext.setComplete();
if (inputs.isEnd()) {
contribution.setExitStatus(ExitStatus.COMPLETED);
}
return RepeatStatus.continueIf(!inputs.isEnd());
}
}

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

@@ -32,6 +32,7 @@ import org.springframework.batch.core.step.skip.SkipListenerFailedException;
import org.springframework.batch.core.step.skip.SkipPolicy;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.retry.ExhaustedRetryException;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
@@ -183,7 +184,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
@SuppressWarnings("unchecked")
UserData<O> data = (UserData<O>) inputs.getUserData();
Chunk<O> cache = data.getOutputs();
final Chunk<O>.ChunkIterator cacheIterator = cache.isEmpty() ? null : cache.iterator();
final Chunk<O>.ChunkIterator cacheIterator = cache.isEmpty() ? null : cache.iterator();
final AtomicInteger count = new AtomicInteger(0);
for (final Chunk<I>.ChunkIterator iterator = inputs.iterator(); iterator.hasNext();) {
@@ -279,7 +280,24 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
if (!inputs.isBusy()) {
chunkMonitor.setChunkSize(inputs.size());
doWrite(outputs.getItems());
try {
doWrite(outputs.getItems());
}
catch (Exception e) {
if (rollbackClassifier.classify(e)) {
throw e;
}
/*
* If the exception is marked as no-rollback, we might
* need to override that if it is also skippable,
* otherwise there's no way to honour the skip listener
* contract.
*/
if (itemWriteSkipPolicy.shouldSkip(e, -1)) {
throw new ForceRollbackForWriteSkipException(
"Force rollback on skippable exception so that skipped item can be located.", e);
}
}
contribution.incrementWriteCount(outputs.size());
}
else {
@@ -332,6 +350,17 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
RecoveryCallback<Object> recoveryCallback = new RecoveryCallback<Object>() {
public Object recover(RetryContext context) throws Exception {
/*
* If the last exception was not skippable we don't need to
* do any scanning. We can just bomb out with a retry
* exhausted.
*/
if (!itemWriteSkipPolicy.shouldSkip(context.getLastThrowable(), -1)) {
throw new ExhaustedRetryException("Retry exhausted after last attempt in recovery path, but exception is not skippable.",
context.getLastThrowable());
}
inputs.setBusy(true);
scan(contribution, inputs, outputs, chunkMonitor);
return null;
@@ -426,12 +455,15 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
Chunk<O>.ChunkIterator outputIterator = outputs.iterator();
List<O> items = Collections.singletonList(outputIterator.next());
inputIterator.next();
try {
writeItems(items);
// If successful we are going to return and allow
// the driver to commit...
doAfterWrite(items);
contribution.incrementWriteCount(1);
inputIterator.remove();
outputIterator.remove();
}
catch (Exception e) {
checkSkipPolicy(inputIterator, outputIterator, e, contribution);
@@ -439,8 +471,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
throw e;
}
}
inputIterator.remove();
outputIterator.remove();
chunkMonitor.incrementOffset();
if (outputs.isEmpty()) {
inputs.setBusy(false);

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

@@ -243,7 +243,25 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
* cause rollback
*/
protected Classifier<Throwable, Boolean> getRollbackClassifier() {
return new BinaryExceptionClassifier(noRollbackExceptionClasses, false);
Classifier<Throwable, Boolean> classifier = new BinaryExceptionClassifier(noRollbackExceptionClasses, false);
// Try to avoid pathological cases where we cannot froce a rollback
// where necessary (should be pretty uncommon):
if (!classifier.classify(new ForceRollbackForWriteSkipException("test", new RuntimeException()))) {
final Classifier<Throwable, Boolean> binary = classifier;
classifier = new Classifier<Throwable, Boolean>() {
public Boolean classify(Throwable classifiable) {
if (ForceRollbackForWriteSkipException.class.isAssignableFrom(classifiable.getClass())) {
return true;
}
return binary.classify(classifiable);
}
};
}
return classifier;
}
/**
@@ -355,6 +373,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
private Collection<Class<? extends Throwable>> getSkippableExceptionClasses() {
HashSet<Class<? extends Throwable>> set = new HashSet<Class<? extends Throwable>>(skippableExceptionClasses);
set.addAll(noRollbackExceptionClasses);
set.add(ForceRollbackForWriteSkipException.class);
return set;
}
@@ -365,7 +384,9 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
if (retryPolicy == null) {
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit);
simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
HashSet<Class<? extends Throwable>> set = new HashSet<Class<? extends Throwable>>(retryableExceptionClasses);
set.add(ForceRollbackForWriteSkipException.class);
simpleRetryPolicy.setRetryableExceptionClasses(set);
retryPolicy = simpleRetryPolicy;
}

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

@@ -0,0 +1,32 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
/**
* Fatal exception to be thrown when a rollback must be forced, typically after
* catching an exception that otherwise would not cause a rollback.
*
* @author Dave Syer
*
*/
public class ForceRollbackForWriteSkipException extends RuntimeException {
public ForceRollbackForWriteSkipException(String msg, Throwable cause) {
super(msg, cause);
}
}

31
spring-batch-core/src/main/java/org/springframework/batch/core/step/skip/NonSkippableWriteException.java Normal file
View File

@@ -0,0 +1,31 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.skip;
/**
* Fatal exception to be thrown when a process operation could not be skipped.
*
* @author Dave Syer
*
*/
public class NonSkippableWriteException extends SkipException {
public NonSkippableWriteException(String msg, Throwable cause) {
super(msg, cause);
}
}

12
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/ChunkElementParserTests.java
View File

@@ -50,7 +50,7 @@ public class ChunkElementParserTests {
public void testInheritSkippable() throws Exception {
Collection<Class<?>> skippable = getExceptionClasses("s1", "skippable",
chunkElementParentAttributeParserTestsContext);
assertEquals(2, skippable.size());
assertEquals(3, skippable.size());
boolean e = false;
boolean f = false;
for (Class<?> cls : skippable) {
@@ -115,10 +115,10 @@ public class ChunkElementParserTests {
}
@Test
public void testInheritSkippable_NoMerge() throws Exception {
public void testInheritSkippableWithNoMerge() throws Exception {
Collection<Class<?>> skippable = getExceptionClasses("s2", "skippable",
chunkElementParentAttributeParserTestsContext);
assertEquals(1, skippable.size());
assertEquals(2, skippable.size());
boolean e = false;
for (Class<?> cls : skippable) {
if (cls.equals(NullPointerException.class)) {
@@ -129,7 +129,7 @@ public class ChunkElementParserTests {
}
@Test
public void testInheritFatal_NoMerge() throws Exception {
public void testInheritFatalWithNoMerge() throws Exception {
Collection<Class<?>> fatal = getExceptionClasses("s2", "fatal", chunkElementParentAttributeParserTestsContext);
boolean a = false;
boolean b = false;
@@ -146,7 +146,7 @@ public class ChunkElementParserTests {
}
@Test
public void testInheritStreams_NoMerge() throws Exception {
public void testInheritStreamsWithNoMerge() throws Exception {
Collection<ItemStream> streams = getStreams("s2", chunkElementParentAttributeParserTestsContext);
assertEquals(1, streams.size());
boolean c = false;
@@ -159,7 +159,7 @@ public class ChunkElementParserTests {
}
@Test
public void testInheritRetryListeners_NoMerge() throws Exception {
public void testInheritRetryListenersWithNoMerge() throws Exception {
Collection<RetryListener> retryListeners = getRetryListeners("s2",
chunkElementParentAttributeParserTestsContext);
assertEquals(1, retryListeners.size());

80
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests.java
View File

@@ -84,17 +84,6 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
assertEquals("[]", writer.getCommitted().toString());
}
@Test
public void testDefaultFatal() throws Exception {
writer.setExceptionType(RuntimeException.class);
StepExecution stepExecution = launchStep("skippableStep");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
// writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[]",
// writer.getCommitted().toString());
}
@Test
public void testSkippable() throws Exception {
writer.setExceptionType(SkippableRuntimeException.class);
@@ -105,10 +94,21 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
}
@Test
public void testFatal() throws Exception {
writer.setExceptionType(FatalRuntimeException.class);
public void testRegularRuntimeExceptionNotSkipped() throws Exception {
writer.setExceptionType(RuntimeException.class);
StepExecution stepExecution = launchStep("skippableStep");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// BATCH-1327:
assertEquals("[1, 2, 3]", writer.getWritten().toString());
// BATCH-1327:
assertEquals("[]", writer.getCommitted().toString());
}
@Test
public void testFatalOverridesSkippable() throws Exception {
writer.setExceptionType(FatalRuntimeException.class);
StepExecution stepExecution = launchStep("skippableFatalStep");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[1, 2, 3]", writer.getWritten().toString());
assertEquals("[]", writer.getCommitted().toString());
}
@@ -116,12 +116,12 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
@Test
public void testDefaultFatalChecked() throws Exception {
writer.setExceptionType(Exception.class);
StepExecution stepExecution = launchStep("skippableStep");
StepExecution stepExecution = launchStep("skippableFatalStep");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
// writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[]",
// writer.getCommitted().toString());
// BATCH-1327:
assertEquals("[1, 2, 3]", writer.getWritten().toString());
// BATCH-1327:
assertEquals("[]", writer.getCommitted().toString());
}
@Test
@@ -136,20 +136,20 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
@Test
public void testFatalChecked() throws Exception {
writer.setExceptionType(FatalException.class);
StepExecution stepExecution = launchStep("skippableStep");
StepExecution stepExecution = launchStep("skippableFatalStep");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[1, 2, 3]", writer.getWritten().toString());
assertEquals("[]", writer.getCommitted().toString());
}
@Test
public void testRetryableDefaultFatal() throws Exception {
public void testRetryableButNotSkippable() throws Exception {
writer.setExceptionType(RuntimeException.class);
StepExecution stepExecution = launchStep("retryable");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[1, 2, 3, 1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[]",
// writer.getCommitted().toString());
assertEquals("[1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
// BATCH-1327:
assertEquals("[]", writer.getCommitted().toString());
}
@Test
@@ -172,13 +172,13 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
}
@Test
public void testRetryableDefaultFatalChecked() throws Exception {
public void testRetryableButNotSkippableChecked() throws Exception {
writer.setExceptionType(Exception.class);
StepExecution stepExecution = launchStep("retryable");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals("[1, 2, 3, 1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[]",
// writer.getCommitted().toString());
assertEquals("[1, 2, 3, 1, 2, 3]", writer.getWritten().toString());
// BATCH-1327:
assertEquals("[]", writer.getCommitted().toString());
}
@Test
@@ -205,10 +205,10 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
writer.setExceptionType(RuntimeException.class);
StepExecution stepExecution = launchStep("noRollbackDefault");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
// writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[]",
// writer.getCommitted().toString());
// BATCH-1318:
assertEquals("[1, 2, 3]", writer.getWritten().toString());
// BATCH-1318:
assertEquals("[]", writer.getCommitted().toString());
}
@Test
@@ -216,8 +216,7 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
writer.setExceptionType(SkippableRuntimeException.class);
StepExecution stepExecution = launchStep("noRollbackDefault");
assertNotNull(stepExecution);
// TODO BATCH-1318: assertEquals(BatchStatus.FAILED,
// stepExecution.getStatus());
// TODO BATCH-1318: assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
// writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
@@ -239,8 +238,9 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
StepExecution stepExecution = launchStep("noRollbackSkippable");
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
assertEquals("[1, 2, 3, 1, 2, 3, 4]", writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[1, 2, 3, 1, 2, 3, 4]",
// writer.getCommitted().toString());
// TODO BATCH-1332: assertEquals("[1, 2, 4]", writer.getCommitted().toString());
// Skipped but also committed!
assertEquals(1, stepExecution.getWriteSkipCount());
}
@Test
@@ -254,12 +254,16 @@ public class FaultTolerantExceptionClassesTests implements ApplicationContextAwa
@Test
public void testNoRollbackFatalNoRollbackException() throws Exception {
// User has asked for no rollback on a fatal exception. What should the
// outcome be?
writer.setExceptionType(FatalRuntimeException.class);
StepExecution stepExecution = launchStep("noRollbackFatal");
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
// TODO BATCH-1331: assertEquals(BatchStatus.FAILED,
// stepExecution.getStatus());
// TODO BATCH-1331: assertEquals("[1, 2, 3]",
// writer.getWritten().toString());
// TODO BATCH-1318: assertEquals("[1, 2, 3]",
// TODO BATCH-1331: assertEquals("[1, 2, 3]",
// writer.getCommitted().toString());
}

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

@@ -471,7 +471,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
// [b]
assertEquals("[b]", provided.toString());
// [b]
assertEquals("[b, b]", processed.toString());
assertEquals("[b]", processed.toString());
// []
assertEquals(0, recovered.size());
assertEquals(1, stepExecution.getReadCount());

11
spring-batch-core/src/test/resources/org/springframework/batch/core/step/item/FaultTolerantExceptionClassesTests-context.xml
View File

@@ -12,6 +12,17 @@
</step>
<step id="skippableStep" xmlns="http://www.springframework.org/schema/batch">
<tasklet>
<chunk reader="reader" writer="writer" commit-interval="4" skip-limit="1">
<skippable-exception-classes>
org.springframework.batch.core.step.item.SkippableRuntimeException
org.springframework.batch.core.step.item.SkippableException
</skippable-exception-classes>
</chunk>
</tasklet>
</step>
<step id="skippableFatalStep" xmlns="http://www.springframework.org/schema/batch">
<tasklet>
<chunk reader="reader" writer="writer" commit-interval="4" skip-limit="1">
<skippable-exception-classes>