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 85bf37d33..ff86e3141 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 @@ -21,8 +21,6 @@ import java.util.Iterator; import java.util.Map; import java.util.Set; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.UnexpectedJobExecutionException; @@ -60,9 +58,7 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler { * 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 int skipCacheCapacity = 1024; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleItemHandler.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleItemHandler.java index adc8d6f33..b6c0c35ba 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleItemHandler.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleItemHandler.java @@ -15,6 +15,8 @@ */ package org.springframework.batch.core.step.item; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.StepContribution; import org.springframework.batch.item.ClearFailedException; import org.springframework.batch.item.FlushFailedException; @@ -38,6 +40,8 @@ import org.springframework.batch.repeat.ExitStatus; */ public class SimpleItemHandler implements ItemHandler { + protected final Log logger = LogFactory.getLog(getClass()); + private ItemReader itemReader; private ItemWriter itemWriter; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index fb2ef4352..48a89d37a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -87,6 +87,22 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { protected Class[] getFatalExceptionClasses() { return fatalExceptionClasses; } + + /** + * Protected getter for the skippable exceptions. + * @return the skippableExceptionClasses + */ + protected Class[] getSkippableExceptionClasses() { + return skippableExceptionClasses; + } + + /** + * Protected getter for the skip limit. + * @return the skipLimit + */ + protected int getSkipLimit() { + return skipLimit; + } /** * Public setter for the {@link ItemKeyGenerator}. This is used to identify diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java index 38d470ba0..4b4562155 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java @@ -15,12 +15,19 @@ */ package org.springframework.batch.core.step.item; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepContribution; -import org.springframework.batch.item.AbstractItemWriter; +import org.springframework.batch.core.listener.CompositeSkipListener; +import org.springframework.batch.core.step.skip.ItemSkipPolicy; +import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; +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.ItemRecoverer; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; @@ -37,12 +44,13 @@ import org.springframework.batch.retry.support.RetryTemplate; /** * Factory bean for step that executes its item processing with a stateful - * retry. Failed items are never skipped, but always cause a rollback. Before a - * rollback, the {@link Step} makes a record of the failed item, caching it - * under a key given by the {@link ItemKeyGenerator}. Then when it is - * re-presented by the {@link ItemReader} it is recognised and retried up to a - * limit given by the {@link RetryPolicy}. When the retry is exhausted instead - * of the item being skipped it is handled by an {@link ItemRecoverer}.
+ * retry. Failed items where the exception is classified as retryable always + * cause a rollback. Before a rollback, the {@link Step} makes a record of the + * failed item, caching it under a key given by the {@link ItemKeyGenerator}. + * Then when it is re-presented by the {@link ItemReader} it is recognised and + * retried up to a limit given by the {@link RetryPolicy}. When the retry is + * exhausted the item is skipped and handled by a {@link SkipListener} if one is + * present.
* * The skipLimit property is still used to control the overall exception * handling policy. Only exhausted retries count against the exception handler, @@ -55,8 +63,6 @@ import org.springframework.batch.retry.support.RetryTemplate; */ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { - private ItemRecoverer itemRecoverer; - private int retryLimit; private Class[] retryableExceptionClasses; @@ -98,18 +104,6 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { this.retryListeners = retryListeners; } - /** - * Public setter for the {@link ItemRecoverer}. If this is set the - * {@link ItemRecoverer#recover(Object, Throwable)} will be called when - * retry is exhausted, and within the business transaction (which will not - * roll back because of any other item-related errors). - * - * @param itemRecoverer the {@link ItemRecoverer} to set - */ - public void setItemRecoverer(ItemRecoverer itemRecoverer) { - this.itemRecoverer = itemRecoverer; - } - /** * @param step * @@ -123,7 +117,8 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { addFatalExceptionIfMissing(RetryException.class); SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(retryLimit); - if (retryableExceptionClasses != null) { + if (retryableExceptionClasses != null) { // otherwise we retry + // all exceptions retryPolicy.setRetryableExceptionClasses(retryableExceptionClasses); retryPolicy.setFatalExceptionClasses(getFatalExceptionClasses()); } @@ -143,9 +138,15 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { retryTemplate.setBackOffPolicy(backOffPolicy); } + List exceptions = new ArrayList(Arrays.asList(getSkippableExceptionClasses())); + if (retryableExceptionClasses != null) { + exceptions.addAll(Arrays.asList(retryableExceptionClasses)); + } + LimitCheckingItemSkipPolicy itemSkipPolicy = new LimitCheckingItemSkipPolicy(getSkipLimit(), exceptions, + Arrays.asList(getFatalExceptionClasses())); StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler(getItemReader(), getItemWriter(), - retryTemplate, getItemKeyGenerator(), itemRecoverer); - itemHandler.setItemSkipPolicy(getItemSkipPolicy()); + retryTemplate, getItemKeyGenerator(), itemSkipPolicy); + itemHandler.setSkipListeners(new BatchListenerFactoryHelper().getSkipListeners(getListeners())); step.setItemHandler(itemHandler); @@ -154,98 +155,139 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { } /** - * Extend the skipping handler because we want to take advantage of that - * behaviour as well as the retry. So if there is an exception on input it - * is skipped if allowed. If there is an exception on output, it will be - * re-thrown in any case, and the behaviour when the item is next - * encountered depends on the retryable and skippable exception - * configuration. Skip takes precedence, so if the exception was skippable - * the item will be skipped on input and the reader moves to the next item. - * If the exception is retryable but not skippable, then the write will be - * attempted up again up to the retry limit. Beyond the retry limit recovery - * takes over. + * If there is an exception on input it is skipped if allowed. If there is + * an exception on output, it will be re-thrown in any case, and the + * behaviour when the item is next encountered depends on the retryable and + * skippable exception configuration. If the exception is retryable the + * write will be attempted again up to the retry limit. When retry attempts + * are exhausted the skip listener is invoked and the skip count + * incremented. A retryable exception is thus also effectively also + * implicitly skippable. * * @author Dave Syer * */ - private static class StatefulRetryItemHandler extends ItemSkipPolicyItemHandler { + private static class StatefulRetryItemHandler extends SimpleItemHandler { final private RetryOperations retryOperations; final private ItemKeyGenerator itemKeyGenerator; - final private ItemRecoverer itemRecoverer; + private CompositeSkipListener listener = new CompositeSkipListener(); + + final private ItemSkipPolicy itemSkipPolicy; /** * @param itemReader * @param itemWriter * @param retryTemplate * @param itemKeyGenerator - * @param itemRecoverer */ public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, - ItemKeyGenerator itemKeyGenerator, ItemRecoverer itemRecoverer) { + ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy itemSkipPolicy) { super(itemReader, itemWriter); this.retryOperations = retryTemplate; this.itemKeyGenerator = itemKeyGenerator; - this.itemRecoverer = itemRecoverer; + this.itemSkipPolicy = itemSkipPolicy; + } + + /** + * 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. + * + * @param listeners + */ + public void setSkipListeners(SkipListener[] listeners) { + for (int i = 0; i < listeners.length; i++) { + registerSkipListener(listeners[i]); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a skip + * process. + * + * @param listener a {@link SkipListener} + */ + public void registerSkipListener(SkipListener listener) { + this.listener.register(listener); + } + + /** + * Tries to read the item from the reader, in case of exception skip the + * item if the skip policy allows, otherwise re-throw. + * + * @param contribution current StepContribution holding skipped items + * count + * @return next item for processing + */ + protected Object read(StepContribution contribution) throws Exception { + + while (true) { + try { + return doRead(); + } + catch (Exception e) { + try { + if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { + // increment skip count and try again + contribution.incrementTemporaryReadSkipCount(); + 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; + } + } + catch (SkipLimitExceededException ex) { + // we are headed for a abnormal ending so bake in the + // skip count + contribution.combineSkipCounts(); + throw ex; + } + } + } + } /** * Execute the business logic, delegating to the writer.
* - * Process the item with the {@link ItemWriter} in a stateful retry. The - * {@link ItemRecoverer} is used (if provided) in the case of an - * exception to apply alternate processing to the item. If the stateful - * retry is in place then the recovery will happen in the next - * transaction automatically, otherwise it might be necessary for - * clients to make the recover method transactional with appropriate - * propagation behaviour (probably REQUIRES_NEW because the call will - * happen in the context of a transaction that is about to rollback).
+ * Process the item with the {@link ItemWriter} in a stateful retry. Any + * {@link SkipListener} provided is called when retry attempts are + * exhausted. The listener callback (on write failure) will happen in + * the next transaction automatically.
* * @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object, * org.springframework.batch.core.StepContribution) */ protected void write(final Object item, final StepContribution contribution) throws Exception { - final ItemWriter writer = new RetryableItemWriter(contribution); RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() { public Object doWithRetry(RetryContext context) throws Throwable { - writer.write(item); + doWrite(item); return null; } }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item); retryCallback.setRecoveryCallback(new RecoveryCallback() { public Object recover(RetryContext context) { - if (itemRecoverer != null) { - return itemRecoverer.recover(item, context.getLastThrowable()); + Throwable t = context.getLastThrowable(); + // TODO: add retryable exceptions as well? (Or ensure that + // all retryable exceptions are skippable?) + if (itemSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { + listener.onSkipInWrite(item, t); } + contribution.incrementWriteSkipCount(); return null; } }); retryOperations.execute(retryCallback); } - /** - * @author Dave Syer - * - */ - private class RetryableItemWriter extends AbstractItemWriter { - - private StepContribution contribution; - - /** - * @param contribution - */ - public RetryableItemWriter(StepContribution contribution) { - this.contribution = contribution; - } - - public void write(Object item) throws Exception { - doWriteWithSkip(item, contribution); - } - - } - } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java index 48a856a56..3665b5817 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java @@ -22,12 +22,16 @@ import java.util.List; import junit.framework.TestCase; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepListener; import org.springframework.batch.core.job.JobSupport; +import org.springframework.batch.core.listener.SkipListenerSupport; import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; @@ -35,9 +39,9 @@ import org.springframework.batch.core.repository.support.SimpleJobRepository; import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.ItemReader; -import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.retry.RetryException; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.transaction.support.TransactionSynchronizationManager; @@ -48,13 +52,15 @@ import org.springframework.transaction.support.TransactionSynchronizationManager */ public class StatefulRetryStepFactoryBeanTests extends TestCase { + protected final Log logger = LogFactory.getLog(getClass()); + private StatefulRetryStepFactoryBean factory = new StatefulRetryStepFactoryBean(); private List recovered = new ArrayList(); private List processed = new ArrayList(); - - int count = 0; + + int count = 0; private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao()); @@ -72,11 +78,11 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { * @see junit.framework.TestCase#setUp() */ protected void setUp() throws Exception { - + MapJobInstanceDao.clear(); MapJobExecutionDao.clear(); MapStepExecutionDao.clear(); - + factory.setBeanName("step"); factory.setItemReader(new ListItemReader(new ArrayList())); @@ -86,10 +92,11 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { JobSupport job = new JobSupport("jobName"); job.setRestartable(true); - JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique").toJobParameters(); + JobParameters jobParameters = new JobParametersBuilder().addString("statefulTest", "make_this_unique") + .toJobParameters(); jobExecution = repository.createJobExecution(job, jobParameters); jobExecution.setEndTime(new Date()); - + } public void testType() throws Exception { @@ -100,14 +107,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { assertTrue(factory.getObject() instanceof Step); } - public void testRecovery() throws Exception { - factory.setItemRecoverer(new ItemRecoverer() { - public Object recover(Object item, Throwable cause) { - recovered.add(item); - assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); - return item; - } - }); + public void testSuccessfulRetry() throws Exception { List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(new String[] { "a", "b", "c" })); ItemReader provider = new ListItemReader(items) { @@ -126,14 +126,15 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); - + assertEquals(0, stepExecution.getSkipCount()); + // b is processed twice, plus 1, plus c, plus the null at end assertEquals(5, count); } - + public void testSkipAndRetry() throws Exception { - factory.setSkippableExceptionClasses(new Class[] {Exception.class}); + factory.setSkippableExceptionClasses(new Class[] { Exception.class }); factory.setSkipLimit(2); List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" })); @@ -153,46 +154,54 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); step.execute(stepExecution); - + assertEquals(2, stepExecution.getSkipCount()); // b is processed once and skipped, plus 1, plus c, plus the null at end assertEquals(7, count); } - - //The following test fails due to current expected behavior of retry - //that will be addressed in 1.1 -// public void testSkipAndRetryWithWriteFailure() throws Exception { -// -// factory.setSkippableExceptionClasses(new Class[] {RetryException.class}); -// factory.setSkipLimit(2); -// List items = TransactionAwareProxyFactory.createTransactionalList(); -// items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" })); -// ItemReader provider = new ListItemReader(items) { -// public Object read() { -// Object item = super.read(); -// System.out.print("Read Called! Item: [" + item + "]"); -// count++; -// return item; -// } -// }; -// -// ItemWriter itemWriter = new AbstractItemWriter(){ -// public void write(Object item) throws Exception { -// System.out.print("Write Called! Item: [" + item + "]"); -// if ("b".equals(item) || "d".equals(item)) { -// throw new RuntimeException("Read error - planned but skippable."); -// } -// }}; -// factory.setItemReader(provider); -// factory.setItemWriter(itemWriter); -// factory.setRetryLimit(5); -// factory.setRetryableExceptionClasses(new Class[]{RuntimeException.class}); -// AbstractStep step = (AbstractStep) factory.getObject(); -// step.setName("mytest"); -// StepExecution stepExecution = new StepExecution(step, jobExecution); -// step.execute(stepExecution); -// -// assertEquals(2, stepExecution.getSkipCount()); -// assertEquals(9, count); -// } + + public void testSkipAndRetryWithWriteFailure() throws Exception { + + factory.setSkippableExceptionClasses(new Class[] { RetryException.class }); + factory.setListeners(new StepListener[] { new SkipListenerSupport() { + public void onSkipInWrite(Object item, Throwable t) { + recovered.add(item); + assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); + } + } }); + factory.setSkipLimit(2); + List items = TransactionAwareProxyFactory.createTransactionalList(); + items.addAll(Arrays.asList(new String[] { "a", "b", "c", "d", "e", "f" })); + ItemReader provider = new ListItemReader(items) { + public Object read() { + Object item = super.read(); + logger.debug("Read Called! Item: [" + item + "]"); + count++; + return item; + } + }; + + ItemWriter itemWriter = new AbstractItemWriter() { + public void write(Object item) throws Exception { + logger.debug("Write Called! Item: [" + item + "]"); + if ("b".equals(item) || "d".equals(item)) { + throw new RuntimeException("Read error - planned but skippable."); + } + } + }; + factory.setItemReader(provider); + factory.setItemWriter(itemWriter); + factory.setRetryLimit(5); + factory.setRetryableExceptionClasses(new Class[] { RuntimeException.class }); + AbstractStep step = (AbstractStep) factory.getObject(); + step.setName("mytest"); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + step.execute(stepExecution); + + assertEquals(2, recovered.size()); + assertEquals(2, stepExecution.getSkipCount()); + assertEquals(2, stepExecution.getWriteSkipCount().intValue()); + // each item once, plus 5 failed retries each for b and d, plus the null terminator + assertEquals(17, count); + } } diff --git a/spring-batch-core/src/test/resources/log4j.properties b/spring-batch-core/src/test/resources/log4j.properties index 857d5f854..f242f56df 100644 --- a/spring-batch-core/src/test/resources/log4j.properties +++ b/spring-batch-core/src/test/resources/log4j.properties @@ -6,6 +6,7 @@ log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - <%m>%n log4j.category.org.apache.activemq=ERROR log4j.category.org.springframework.batch=DEBUG +log4j.category.org.springframework.batch.support=INFO # log4j.category.org.springframework.transaction=INFO # log4j.category.org.hibernate.SQL=DEBUG