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 9bd7fcd63..fc46d4305 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 @@ -16,7 +16,10 @@ package org.springframework.batch.core.step.item; import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; import java.util.Map; +import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -28,6 +31,8 @@ import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.MarkFailedException; +import org.springframework.transaction.support.TransactionSynchronizationManager; /** * {@link ItemHandler} that implements skip behavior. It delegates to @@ -42,10 +47,17 @@ import org.springframework.batch.item.ItemWriter; */ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { + /** + * 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"; + protected final Log logger = LogFactory.getLog(getClass()); private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy(); + private Map skippedExceptions = new HashMap(); + private ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() { public Object getKey(Object item) { return item; @@ -56,8 +68,6 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { private CompositeSkipListener listener = new CompositeSkipListener(); - private Map skippedExceptions = new HashMap(); - /** * Register some {@link SkipListener}s with the handler. Each will get the * callbacks in the order specified at the correct stage if a skip occurs. @@ -128,13 +138,16 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { Object item = doRead(); Object key = itemKeyGenerator.getKey(item); - while (item != null && skippedExceptions.containsKey(key)) { + Throwable throwable = getSkippedException(key); + while (item != null && throwable != null) { logger.debug("Skipping item on input, previously failed on output; key=[" + key + "]"); + scheduleForRemoval(key); if (listener != null) { - listener.onSkipInWrite(item, (Throwable) skippedExceptions.get(key)); + listener.onSkipInWrite(item, throwable); } item = doRead(); key = itemKeyGenerator.getKey(item); + throwable = getSkippedException(key); } return item; @@ -178,12 +191,66 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { contribution.incrementSkipCount(); // don't call the listener here - the transaction is going to // roll back - skippedExceptions.put(key, e); + addSkippedException(key, e); logger.debug("Added item to skip list; key=" + key); } // always re-throw exception on write throw e; } } + + public void mark() throws MarkFailedException { + super.mark(); + clearSkippedExceptions(); + } + + /** + * Synchronized setter for a skipped exception. + */ + private void addSkippedException(Object key, Throwable e) { + synchronized (skippedExceptions) { + skippedExceptions.put(key, e); + } + } + + /** + * Schedule this key for removal from the skipped exception cache at the end + * of this transaction (only removed if business transaction is successful). + * + * @param key + */ + private void scheduleForRemoval(Object key) { + if (!TransactionSynchronizationManager.hasResource(TO_BE_REMOVED)) { + TransactionSynchronizationManager.bindResource(TO_BE_REMOVED, new HashSet()); + } + ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).add(key); + } + + /** + * Clear the map of skipped exception corresponding to key. + * @param key the key to clear + */ + private void clearSkippedExceptions() { + if (!TransactionSynchronizationManager.hasResource(TO_BE_REMOVED)) { + return; + } + synchronized (skippedExceptions) { + for (Iterator iterator = ((Set) TransactionSynchronizationManager.getResource(TO_BE_REMOVED)).iterator(); iterator.hasNext();) { + Object key = (Object) iterator.next(); + skippedExceptions.remove(key); + } + TransactionSynchronizationManager.unbindResource(TO_BE_REMOVED); + } + } + + /** + * Synchronized getter for a skipped exception. + * @return the skippedExceptions + */ + private Throwable getSkippedException(Object key) { + synchronized (skippedExceptions) { + return (Throwable) skippedExceptions.get(key); + } + } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java index 500b4d953..e10dd86bf 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemSkipPolicyItemHandlerTests.java @@ -18,6 +18,7 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Set; import junit.framework.TestCase; @@ -30,6 +31,7 @@ import org.springframework.batch.core.step.StepSupport; import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy; import org.springframework.batch.item.ClearFailedException; import org.springframework.batch.item.FlushFailedException; +import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.MarkFailedException; @@ -37,6 +39,7 @@ import org.springframework.batch.item.NoWorkFoundException; import org.springframework.batch.item.ParseException; import org.springframework.batch.item.ResetFailedException; import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.transaction.support.TransactionSynchronizationManager; /** * @author Dave Syer @@ -44,12 +47,18 @@ import org.springframework.batch.item.UnexpectedInputException; */ public class ItemSkipPolicyItemHandlerTests extends TestCase { - private ItemSkipPolicyItemHandler handler = new ItemSkipPolicyItemHandler(new SkipReaderStub(), - new SkipWriterStub()); + private final SkipWriterStub writer = new SkipWriterStub(); + + private ItemSkipPolicyItemHandler handler = new ItemSkipPolicyItemHandler(new SkipReaderStub(), writer); private StepContribution contribution = new StepContribution(new JobExecution(new JobInstance(new Long(11), new JobParameters(), new JobSupport())).createStepExecution(new StepSupport("foo"))); + protected void tearDown() throws Exception { + // remove the resource if it exists + handler.mark(); + } + public void testReadWithNoSkip() throws Exception { assertEquals(new Holder("1"), handler.read(contribution)); try { @@ -83,8 +92,57 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { assertEquals(0, contribution.getContributionSkipCount()); } - public void testWriteWithSkip() throws Exception { + public void testHandleWithSkip() throws Exception { handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + handler.handle(contribution); + handler.handle(contribution); + assertEquals(1, contribution.getContributionSkipCount()); + // 2 is skipped so 3 was last one processed and now we are at 4 + try { + handler.handle(contribution); + fail("Expected SkippableException"); + } + catch (SkippableException e) { + // expected + } + assertEquals(2, contribution.getContributionSkipCount()); + // No "4" because it was skipped on write + assertEquals(new Holder("5"), handler.read(contribution)); + } + + public void testWriteWithSkipAfterMark() throws Exception { + handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + try { + handler.write(new Holder("4"), contribution); + fail("Expected SkippableException"); + } + catch (SkippableException e) { + // expected + } + handler.handle(contribution); + handler.handle(contribution); + assertEquals(2, contribution.getContributionSkipCount()); + // 2 is skipped so 3 was last one processed and now we are at 4, which was previously skipped + handler.handle(contribution); + assertEquals(null, handler.read(contribution)); + assertEquals(2, contribution.getContributionSkipCount()); + + assertEquals(1, TransactionSynchronizationManager.getResourceMap().size()); + Set removed = (Set) TransactionSynchronizationManager.getResourceMap().values().iterator().next(); + // one skipped item was detected on read + assertEquals(1, removed.size()); + // mark() should remove the set of removed keys + handler.mark(); + assertEquals(0, TransactionSynchronizationManager.getResourceMap().size()); + } + + public void testWriteWithSkipAndItemKeyGenerator() throws Exception { + handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + handler.setItemKeyGenerator(new ItemKeyGenerator() { + public Object getKey(Object item) { + return ((Holder) item).value; + } + }); handler.write(new Holder("3"), contribution); try { handler.write(new Holder("4"), contribution); @@ -94,17 +152,40 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { // expected } assertEquals(1, contribution.getContributionSkipCount()); + assertEquals(new Holder("1"), handler.read(contribution)); + assertEquals(new Holder("3"), handler.read(contribution)); + assertEquals(2, contribution.getContributionSkipCount()); + // No "4" because it was skipped on write + assertEquals(new Holder("5"), handler.read(contribution)); + } + + public void testWriteWithSkipWhenMutating() throws Exception { + handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + writer.mutate = true; + handler.setItemSkipPolicy(new AlwaysSkipItemSkipPolicy()); + handler.handle(contribution); + handler.handle(contribution); + assertEquals(1, contribution.getContributionSkipCount()); + // 2 is skipped so 3 was last one processed and now we are at 4 + try { + handler.handle(contribution); + fail("Expected SkippableException"); + } + catch (SkippableException e) { + // expected + } + assertEquals(2, contribution.getContributionSkipCount()); + // No "4" because it was skipped on write, even though it is mutating + // its key + assertEquals(new Holder("5"), handler.read(contribution)); } - - // TODO: test the item key generator, especially with a mutable item that changes on write /** * Simple item reader that supports skip functionality. */ private static class SkipReaderStub implements ItemReader { - final Holder[] items = { new Holder("1"), new Holder("2"), new Holder("3"), new Holder("4"), new Holder("5"), - null }; + final String[] values = { "1", "2", "3", "4", "5" }; Collection processed = new ArrayList(); @@ -117,8 +198,12 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { if (counter == 1) { throw new SkippableException("exception in reader"); } - processed.add(items[counter]); - return items[counter]; + if (counter >= values.length) { + return null; + } + Object item = new Holder(values[counter]); + processed.add(item); + return item; } public void mark() throws MarkFailedException { @@ -136,6 +221,8 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { */ private static class SkipWriterStub implements ItemWriter { + boolean mutate = false; + List written = new ArrayList(); int flushIndex = -1; @@ -151,8 +238,12 @@ public class ItemSkipPolicyItemHandlerTests extends TestCase { } public void write(Object item) throws Exception { + String value = ((Holder) item).value; written.add(item); - if (((Holder) item).value.equals("4")) { + if (mutate) { + ((Holder) item).value = "done"; + } + if (value.equals("4")) { throw new SkippableException("exception in writer"); } }