From f72cd89bc62bca9ff1001fcd5bf686afc3c449fb Mon Sep 17 00:00:00 2001 From: robokaso Date: Tue, 18 Nov 2008 10:08:23 +0000 Subject: [PATCH] RESOLVED - BATCH-926: vague skip limit for concurrent chunks documented status quo --- .../step/item/SkipLimitStepFactoryBean.java | 149 ++++++++---------- 1 file changed, 63 insertions(+), 86 deletions(-) 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 aa7a360f4..41e02112e 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 @@ -78,8 +78,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * properties are ignored (retryLimit, backOffPolicy, * retryableExceptionClasses). * - * @param retryPolicy - * a stateless {@link RetryPolicy} + * @param retryPolicy a stateless {@link RetryPolicy} */ public void setRetryPolicy(RetryPolicy retryPolicy) { this.retryPolicy = retryPolicy; @@ -89,8 +88,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * Public setter for the retry limit. Each item can be retried up to this * limit. * - * @param retryLimit - * the retry limit to set + * @param retryLimit the retry limit to set */ public void setRetryLimit(int retryLimit) { this.retryLimit = retryLimit; @@ -111,8 +109,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * this many failures in a single transaction. Defaults to the value in the * {@link MapRetryContextCache}. * - * @param cacheCapacity - * the cacheCapacity to set + * @param cacheCapacity the cacheCapacity to set */ public void setCacheCapacity(int cacheCapacity) { this.cacheCapacity = cacheCapacity; @@ -121,8 +118,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { /** * Public setter for the Class[]. * - * @param retryableExceptionClasses - * the retryableExceptionClasses to set + * @param retryableExceptionClasses the retryableExceptionClasses to set */ public void setRetryableExceptionClasses(Class[] retryableExceptionClasses) { this.retryableExceptionClasses = retryableExceptionClasses; @@ -131,8 +127,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { /** * Public setter for the {@link BackOffPolicy}. * - * @param backOffPolicy - * the {@link BackOffPolicy} to set + * @param backOffPolicy the {@link BackOffPolicy} to set */ public void setBackOffPolicy(BackOffPolicy backOffPolicy) { this.backOffPolicy = backOffPolicy; @@ -141,8 +136,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { /** * Public setter for the {@link RetryListener}s. * - * @param retryListeners - * the {@link RetryListener}s to set + * @param retryListeners the {@link RetryListener}s to set */ public void setRetryListeners(RetryListener[] retryListeners) { this.retryListeners = retryListeners; @@ -155,8 +149,13 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * zero then all exceptions will be propagated from the chunk and cause the * step to abort. * - * @param skipLimit - * the value to set. Default is 0 (never skip). + * Note that if chunks are executed concurrently the number of skips can + * potentially exceed the skip limit and step can still finish successfully. + * This is due to the fact that overall skip count is not being synchronized + * between concurrent chunks while they processing, only on chunk + * boundaries. + * + * @param skipLimit the value to set. Default is 0 (never skip). */ public void setSkipLimit(int skipLimit) { this.skipLimit = skipLimit; @@ -167,8 +166,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * but will result in transaction rollback and the item which handling * caused the exception will be skipped. * - * @param exceptionClasses - * defaults to Exception + * @param exceptionClasses defaults to Exception */ public void setSkippableExceptionClasses(Class[] exceptionClasses) { this.skippableExceptionClasses = exceptionClasses; @@ -177,8 +175,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { /** * Public setter for exception classes that should cause immediate failure. * - * @param fatalExceptionClasses - * {@link Error} by default + * @param fatalExceptionClasses {@link Error} by default */ public void setFatalExceptionClasses(Class[] fatalExceptionClasses) { this.fatalExceptionClasses = fatalExceptionClasses; @@ -189,8 +186,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * failed items so they can be skipped if encountered again, generally in * another transaction. * - * @param itemKeyGenerator - * the {@link ItemKeyGenerator} to set. + * @param itemKeyGenerator the {@link ItemKeyGenerator} to set. */ public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) { this.itemKeyGenerator = itemKeyGenerator; @@ -211,16 +207,13 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { if (retryPolicy == null) { - SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy( - retryLimit); + SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(retryLimit); if (retryableExceptionClasses.length > 0) { // otherwise we - // retry + // retry // all exceptions - simpleRetryPolicy - .setRetryableExceptionClasses(retryableExceptionClasses); + simpleRetryPolicy.setRetryableExceptionClasses(retryableExceptionClasses); } - simpleRetryPolicy - .setFatalExceptionClasses(fatalExceptionClasses); + simpleRetryPolicy.setFatalExceptionClasses(fatalExceptionClasses); ExceptionClassifierRetryPolicy classifierRetryPolicy = new ExceptionClassifierRetryPolicy(); SubclassExceptionClassifier exceptionClassifier = new SubclassExceptionClassifier(); @@ -234,27 +227,22 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { retryPolicyMap.put("retry", simpleRetryPolicy); retryPolicyMap.put("default", new NeverRetryPolicy()); classifierRetryPolicy.setPolicyMap(retryPolicyMap); - classifierRetryPolicy - .setExceptionClassifier(exceptionClassifier); + classifierRetryPolicy.setExceptionClassifier(exceptionClassifier); retryPolicy = classifierRetryPolicy; } // Co-ordinate the retry policy with the exception handler: getStepOperations().setExceptionHandler( - new SimpleRetryExceptionHandler(retryPolicy, - getExceptionHandler(), fatalExceptionClasses)); + new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), fatalExceptionClasses)); - RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy( - retryPolicy) { + RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) { protected boolean recoverForException(Throwable ex) { return !getTransactionAttribute().rollbackOn(ex); } }; if (cacheCapacity > 0) { - recoveryCallbackRetryPolicy - .setRetryContextCache(new MapRetryContextCache( - cacheCapacity)); + recoveryCallbackRetryPolicy.setRetryContextCache(new MapRetryContextCache(cacheCapacity)); } RetryTemplate retryTemplate = new RetryTemplate(); @@ -266,37 +254,32 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { retryTemplate.setBackOffPolicy(backOffPolicy); } - List exceptions = new ArrayList(Arrays - .asList(skippableExceptionClasses)); - ItemSkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy( - skipLimit, exceptions, Arrays.asList(fatalExceptionClasses)); + List exceptions = new ArrayList(Arrays.asList(skippableExceptionClasses)); + ItemSkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, Arrays + .asList(fatalExceptionClasses)); exceptions.addAll(Arrays.asList(retryableExceptionClasses)); - ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy( - skipLimit, exceptions, Arrays.asList(fatalExceptionClasses)); - StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler( - getItemReader(), getItemWriter(), retryTemplate, - itemKeyGenerator, readSkipPolicy, writeSkipPolicy); - itemHandler.setSkipListeners(BatchListenerFactoryHelper - .getSkipListeners(getListeners())); + ItemSkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, exceptions, Arrays + .asList(fatalExceptionClasses)); + StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler(getItemReader(), getItemWriter(), + retryTemplate, itemKeyGenerator, readSkipPolicy, writeSkipPolicy); + itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners())); step.setItemHandler(itemHandler); - } else { + } + else { // This is the default in ItemOrientedStep anyway... - step.setItemHandler(new SimpleItemHandler(getItemReader(), - getItemWriter())); + step.setItemHandler(new SimpleItemHandler(getItemReader(), getItemWriter())); } } public void addFatalExceptionIfMissing(Class cls) { - List fatalExceptionList = new ArrayList(Arrays - .asList(fatalExceptionClasses)); + List fatalExceptionList = new ArrayList(Arrays.asList(fatalExceptionClasses)); if (!fatalExceptionList.contains(cls)) { fatalExceptionList.add(cls); } - fatalExceptionClasses = (Class[]) fatalExceptionList - .toArray(new Class[0]); + fatalExceptionClasses = (Class[]) fatalExceptionList.toArray(new Class[0]); } /** @@ -330,10 +313,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * @param retryTemplate * @param itemKeyGenerator */ - public StatefulRetryItemHandler(ItemReader itemReader, - ItemWriter itemWriter, RetryOperations retryTemplate, - ItemKeyGenerator itemKeyGenerator, - ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) { + public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, + ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) { super(itemReader, itemWriter); this.retryOperations = retryTemplate; this.itemKeyGenerator = itemKeyGenerator; @@ -358,8 +339,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * Register a listener for callbacks at the appropriate stages in a skip * process. * - * @param listener - * a {@link SkipListener} + * @param listener a {@link SkipListener} */ public void registerSkipListener(SkipListener listener) { this.listener.register(listener); @@ -369,8 +349,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * 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 + * @param contribution current StepContribution holding skipped items + * count * @return next item for processing */ protected Object read(StepContribution contribution) throws Exception { @@ -378,18 +358,20 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { while (true) { try { return doRead(); - } catch (Exception e) { + } + catch (Exception e) { try { - if (readSkipPolicy.shouldSkip(e, contribution - .getStepSkipCount())) { + if (readSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) { // increment skip count and try again contribution.incrementTemporaryReadSkipCount(); onSkipInRead(e); logger.debug("Skipping failed input", e); - } else { + } + else { throw new NonSkippableException("Non-skippable exception during read", e); } - } catch (SkipLimitExceededException ex) { + } + catch (SkipLimitExceededException ex) { // we are headed for a abnormal ending so bake in the // skip count contribution.combineSkipCounts(); @@ -409,24 +391,19 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { * the next transaction automatically.
* * @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object, - * org.springframework.batch.core.StepContribution) + * org.springframework.batch.core.StepContribution) */ - protected void write(final Object item, - final StepContribution contribution) throws Exception { - RecoveryRetryCallback retryCallback = new RecoveryRetryCallback( - item, new RetryCallback() { - public Object doWithRetry(RetryContext context) - throws Throwable { - doWrite(item); - return null; - } - }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) - : item); + protected void write(final Object item, final StepContribution contribution) throws Exception { + RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + doWrite(item); + return null; + } + }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item); retryCallback.setRecoveryCallback(new RecoveryCallback() { public Object recover(RetryContext context) { Throwable t = context.getLastThrowable(); - if (writeSkipPolicy.shouldSkip(t, contribution - .getStepSkipCount())) { + if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) { listener.onSkipInWrite(item, t); } else { @@ -438,13 +415,13 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean { }); retryOperations.execute(retryCallback); } - - private void onSkipInRead(Exception e){ - - try{ + + private void onSkipInRead(Exception e) { + + try { listener.onSkipInRead(e); } - catch(Exception ex){ + catch (Exception ex) { logger.debug("Error in SkipListener onSkipInReader encountered and ignored.", ex); } }