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

REOPENED - issue BATCH-511: read failures cause rollback

Browse Source
This commit is contained in:
dsyer
2008-03-27 10:30:06 +00:00
parent 419fa9794d
commit e46c972d8e
4 changed files with 125 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java
View File

@@ -32,6 +32,8 @@ public class StepContribution {
private int skipCount;
private int readSkipCount;
/**
* @param execution
*/
@@ -72,10 +74,11 @@ public class StepContribution {
/**
* @return the sum of skips accumulated in the parent {@link StepExecution}
* and this <code>StepContribution</code>.
* and this <code>StepContribution</code>, including uncommitted read
* skips.
*/
public int getStepSkipCount() {
return skipCount + parentSkipCount;
return readSkipCount + skipCount + parentSkipCount;
}
/**
@@ -88,18 +91,25 @@ public class StepContribution {
}
/**
* Increment the skip count
* Increment the total skip count for this contribution
*/
public void incrementSkipCount() {
skipCount++;
}
/**
* Increment the skip count by a non-trivial amount
* @param delta
* Increment the counter for skipped reads
*/
public void incrementSkipCount(int delta) {
skipCount += delta;
public void incrementReadSkipCount() {
this.readSkipCount++;
}
/**
* Clear the read skips and transfer them to the total skip count
*/
public void commitReadSkipCount() {
skipCount += readSkipCount;
readSkipCount = 0;
}
/*
@@ -107,7 +117,8 @@ public class StepContribution {
* @see java.lang.Object#toString()
*/
public String toString() {
return "[StepContribution: items="+itemCount+", commits=" + commitCount + ", skips=" + skipCount + "]";
return "[StepContribution: items=" + itemCount + ", commits=" + commitCount + ", readSkips=" + readSkipCount
+ ", skips=" + skipCount + "]";
}
}

12
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java
View File

@@ -299,6 +299,7 @@ public class ItemOrientedStep extends AbstractStep {
Thread.currentThread().interrupt();
}
contribution.commitReadSkipCount();
// Apply the contribution to the step
// only if chunk was successful
stepExecution.apply(contribution);
@@ -330,13 +331,11 @@ public class ItemOrientedStep extends AbstractStep {
}
catch (Error e) {
stepExecution.incrementSkipCountBy(contribution.getSkipCount());
processRollback(stepExecution, fatalException, transaction);
processRollback(stepExecution, contribution, fatalException, transaction);
throw e;
}
catch (Exception e) {
stepExecution.incrementSkipCountBy(contribution.getSkipCount());
processRollback(stepExecution, fatalException, transaction);
processRollback(stepExecution, contribution, fatalException, transaction);
throw e;
}
finally {
@@ -525,11 +524,14 @@ public class ItemOrientedStep extends AbstractStep {
/**
* @param stepExecution
* @param contribution
* @param fatalException
* @param transaction
*/
private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException,
private void processRollback(final StepExecution stepExecution, final StepContribution contribution, final ExceptionHolder fatalException,
TransactionStatus transaction) {
stepExecution.incrementSkipCountBy(contribution.getSkipCount());
/*
* Any exception thrown within the transaction should automatically
* cause the transaction to rollback.

40
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java
View File

@@ -28,6 +28,7 @@ import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.listener.CompositeSkipListener;
import org.springframework.batch.core.step.skip.ItemSkipPolicy;
import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
import org.springframework.batch.core.step.skip.SkipLimitExceededException;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
@@ -48,9 +49,10 @@ import org.springframework.transaction.support.TransactionSynchronizationManager
public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
/**
* Key for transaction resource that holds skipped keys until they can be removed
* Key for transaction resource that holds skipped keys until they can be
* removed
*/
private static final String TO_BE_REMOVED = ItemSkipPolicyItemHandler.class.getName()+".TO_BE_REMOVED";
private static final String TO_BE_REMOVED = ItemSkipPolicyItemHandler.class.getName() + ".TO_BE_REMOVED";
protected final Log logger = LogFactory.getLog(getClass());
@@ -153,17 +155,26 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
}
catch (Exception e) {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementSkipCount();
if (listener != null) {
listener.onSkipInRead(e);
try {
if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
// increment skip count and try again
contribution.incrementReadSkipCount();
if (listener != null) {
listener.onSkipInRead(e);
}
logger.debug("Skipping failed input", e);
}
else {
// re-throw only when the skip policy runs out of
// patience
throw e;
}
logger.debug("Skipping failed input", e);
}
else {
// re-throw only when the skip policy runs out of patience
throw e;
catch (SkipLimitExceededException ex) {
// we are headed for a abnormal ending so bake in the skip
// count
contribution.commitReadSkipCount();
throw ex;
}
}
@@ -198,7 +209,7 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
throw e;
}
}
public void mark() throws MarkFailedException {
super.mark();
clearSkippedExceptions();
@@ -235,9 +246,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
return;
}
synchronized (skippedExceptions) {
for (Iterator iterator = ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).iterator(); iterator.hasNext();) {
for (Iterator iterator = ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).iterator(); iterator
.hasNext();) {
Object key = (Object) iterator.next();
skippedExceptions.remove(key);
skippedExceptions.remove(key);
}
TransactionSynchronizationManager.unbindResource(TO_BE_REMOVED);
}

79
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java
View File

@@ -140,10 +140,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
factory.setSkipLimit(3);
factory.setItemReader(reader);
factory.setSkippableExceptionClasses(new Class[] {Exception.class});
factory.setSkippableExceptionClasses(new Class[] { Exception.class });
ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
@@ -161,8 +161,62 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
// writer did not skip "2" as it never made it to writer, only "4" did
assertTrue(reader.processed.contains("4"));
// failure on "4" tripped the skip limit so we never got to "5"
List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1"));
// failure on "4" tripped the skip limit so we never write anything
// ("1" was written but rolled back)
List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
assertEquals(expectedOutput, writer.written);
}
/**
* Check items causing errors are skipped as expected.
*/
public void testSkipOnReadNotDoubleCounted() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6"), StringUtils
.commaDelimitedListToSet("2,3,5"));
factory.setSkipLimit(4);
factory.setItemReader(reader);
ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
StepExecution stepExecution = jobExecution.createStepExecution(step);
step.execute(stepExecution);
assertEquals(4, stepExecution.getSkipCount());
// skipped 2,3,4,5
List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6"));
assertEquals(expectedOutput, writer.written);
}
/**
* Check items causing errors are skipped as expected.
*/
public void testSkipOnWriteNotDoubleCounted() throws Exception {
reader = new SkipReaderStub(StringUtils.commaDelimitedListToStringArray("1,2,3,4,5,6,7"), StringUtils
.commaDelimitedListToSet("2,3"));
writer = new SkipWriterStub(StringUtils.commaDelimitedListToSet("4,5"));
factory.setSkipLimit(4);
factory.setItemReader(reader);
factory.setItemWriter(writer);
ItemOrientedStep step = (ItemOrientedStep) factory.getObject();
StepExecution stepExecution = jobExecution.createStepExecution(step);
step.execute(stepExecution);
System.err.println(writer.written);
System.err.println(reader.processed);
assertEquals(4, stepExecution.getSkipCount());
// skipped 2,3,4,5
List expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("1,6,7"));
assertEquals(expectedOutput, writer.written);
}
@@ -223,6 +277,19 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
int flushIndex = -1;
private final Collection failures;
public SkipWriterStub() {
this(StringUtils.commaDelimitedListToSet("4"));
}
/**
* @param commaDelimitedListToSet
*/
public SkipWriterStub(Collection failures) {
this.failures = failures;
}
public void clear() throws ClearFailedException {
for (int i = flushIndex + 1; i < written.size(); i++) {
written.remove(i);
@@ -234,10 +301,10 @@ public class SkipLimitStepFactoryBeanTests extends TestCase {
}
public void write(Object item) throws Exception {
written.add(item);
if (item.equals("4")) {
if (failures.contains(item)) {
throw new SkippableRuntimeException("exception in writer");
}
written.add(item);
}
}