From e46c972d8e8a267061a7e9a3886cfad32b2f219a Mon Sep 17 00:00:00 2001 From: dsyer Date: Thu, 27 Mar 2008 10:30:06 +0000 Subject: [PATCH] REOPENED - issue BATCH-511: read failures cause rollback --- .../batch/core/StepContribution.java | 27 +++++-- .../core/step/item/ItemOrientedStep.java | 12 +-- .../step/item/ItemSkipPolicyItemHandler.java | 40 ++++++---- .../item/SkipLimitStepFactoryBeanTests.java | 79 +++++++++++++++++-- 4 files changed, 125 insertions(+), 33 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java index 922c3e7c4..f589a18c6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java @@ -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 StepContribution. + * and this StepContribution, 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 + "]"; } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java index 0190f5f91..acfce1a38 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStep.java @@ -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. diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java index fc46d4305..d4b0cf1db 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandler.java @@ -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); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java index 91f7b16a9..43cb5a394 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBeanTests.java @@ -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); } }