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 9f2b46de3..b6d7d307d 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 @@ -22,13 +22,16 @@ 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; +import org.springframework.batch.retry.RetryContext; import org.springframework.batch.retry.RetryException; import org.springframework.batch.retry.RetryListener; import org.springframework.batch.retry.RetryOperations; import org.springframework.batch.retry.RetryPolicy; import org.springframework.batch.retry.backoff.BackOffPolicy; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; -import org.springframework.batch.retry.policy.ItemWriterRetryPolicy; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; import org.springframework.batch.retry.policy.SimpleRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; @@ -126,16 +129,16 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { } // Co-ordinate the retry policy with the exception handler: - getStepOperations() - .setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), getFatalExceptionClasses())); + getStepOperations().setExceptionHandler( + new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), getFatalExceptionClasses())); - ItemWriterRetryPolicy itemWriterRetryPolicy = new ItemWriterRetryPolicy(retryPolicy); + RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy); RetryTemplate retryTemplate = new RetryTemplate(); if (retryListeners != null) { retryTemplate.setListeners(retryListeners); } - retryTemplate.setRetryPolicy(itemWriterRetryPolicy); + retryTemplate.setRetryPolicy(recoveryCallbackRetryPolicy); if (backOffPolicy != null) { retryTemplate.setBackOffPolicy(backOffPolicy); } @@ -203,11 +206,22 @@ public class StatefulRetryStepFactoryBean extends SkipLimitStepFactoryBean { * @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object, * org.springframework.batch.core.StepContribution) */ - protected void write(Object item, final StepContribution contribution) throws Exception { - ItemWriter writer = new RetryableItemWriter(contribution); - ItemWriterRetryCallback retryCallback = new ItemWriterRetryCallback(item, writer); - retryCallback.setKeyGenerator(itemKeyGenerator); - retryCallback.setRecoverer(itemRecoverer); + 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); + return null; + } + }, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item); + retryCallback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable throwable) { + if (itemRecoverer != null) { + return itemRecoverer.recover(item, throwable); + } + return null; + } + }); retryOperations.execute(retryCallback); } 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 d7a024964..48a856a56 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 @@ -102,10 +102,10 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase { public void testRecovery() throws Exception { factory.setItemRecoverer(new ItemRecoverer() { - public boolean recover(Object item, Throwable cause) { + public Object recover(Object item, Throwable cause) { recovered.add(item); assertTrue(TransactionSynchronizationManager.isActualTransactionActive()); - return true; + return item; } }); List items = TransactionAwareProxyFactory.createTransactionalList(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java index 4f3e54f50..7844745f6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java @@ -35,5 +35,5 @@ public interface ItemRecoverer { * the cause of the failure that led to this recovery. * @return true if recovery was successful. */ - boolean recover(Object data, Throwable cause); + Object recover(Object data, Throwable cause); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/NewItemIdentifier.java similarity index 69% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/NewItemIdentifier.java index 7f6372f02..9069dc81a 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/NewItemIdentifier.java @@ -18,21 +18,20 @@ package org.springframework.batch.item; /** * Strategy interface to distinguish a new item from one that has been processed - * before and failed, e.g. by examining a message flag. + * before and one that has not, e.g. by examining a message flag. * * @author Dave Syer * */ -public interface FailedItemIdentifier { +public interface NewItemIdentifier { /** - * Inspect the item and determine if it has previously failed processing. - * The safest choice when the answer is indeterminate is 'true'. + * Inspect the item and determine if it has never been processed before. + * The safest choice when the answer is indeterminate is 'false'. * * @param item the current item. - * @return true if the item has been seen before and is known to have failed - * processing. + * @return true if the item is known to have never been processed before. */ - boolean hasFailed(Object item); + boolean isNew(Object item); } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java index b9372f84a..edc7f50e8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java @@ -23,7 +23,7 @@ import javax.jms.Message; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.item.AbstractItemReader; -import org.springframework.batch.item.FailedItemIdentifier; +import org.springframework.batch.item.NewItemIdentifier; import org.springframework.batch.item.ItemKeyGenerator; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemRecoverer; @@ -37,12 +37,12 @@ import org.springframework.util.Assert; * An {@link ItemReader} for JMS using a {@link JmsTemplate}. The template * should have a default destination, which will be used to provide items in * {@link #read()}. If a recovery step is needed, set the error destination and - * the item will be sent there if processing fails in an external retry. + * the item will be sent there if processing fails in a stateful retry. * * @author Dave Syer * */ -public class JmsItemReader extends AbstractItemReader implements ItemRecoverer, ItemKeyGenerator, FailedItemIdentifier { +public class JmsItemReader extends AbstractItemReader implements ItemRecoverer, ItemKeyGenerator, NewItemIdentifier { protected Log logger = LogFactory.getLog(getClass()); @@ -108,13 +108,14 @@ public class JmsItemReader extends AbstractItemReader implements ItemRecoverer, } /** - * Send the message back to the proovider using the specified error - * destination property of this provider. + * Send the message back to the provider using the specified error + * destination property of this reader. If the recovery is successful the + * item itself is returned, otherwise null. * * @see org.springframework.batch.item.ItemRecoverer#recover(Object, * Throwable) */ - public boolean recover(Object item, Throwable cause) { + public Object recover(Object item, Throwable cause) { try { if (errorDestination != null) { jmsTemplate.convertAndSend(errorDestination, item); @@ -124,15 +125,14 @@ public class JmsItemReader extends AbstractItemReader implements ItemRecoverer, } else { // do nothing - it doesn't make sense to send the message back - // to - // the destination it came from - return false; + // to the destination it came from + return null; } - return true; + return item; } catch (JmsException e) { logger.error("Could not recover because of JmsException.", e); - return false; + throw e; } } @@ -158,21 +158,21 @@ public class JmsItemReader extends AbstractItemReader implements ItemRecoverer, } /** - * If the item is a message, check the JMS redelivered flag, otherwise - * return true to be on the safe side. + * If the item is a message, check the JMS re-delivered flag, otherwise + * return false to be on the safe side. * - * @see org.springframework.batch.item.FailedItemIdentifier#hasFailed(java.lang.Object) + * @see org.springframework.batch.item.NewItemIdentifier#isNew(java.lang.Object) */ - public boolean hasFailed(Object item) { + public boolean isNew(Object item) { if (itemType != null && itemType.isAssignableFrom(Message.class)) { try { - return ((Message) item).getJMSRedelivered(); + return !((Message) item).getJMSRedelivered(); } catch (JMSException e) { throw new UnexpectedInputException("Could not extract message ID", e); } } - return true; + return false; } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/RepeatOperationsInterceptor.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/RepeatOperationsInterceptor.java index 85f537ae5..a96ce31e3 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/RepeatOperationsInterceptor.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/RepeatOperationsInterceptor.java @@ -16,6 +16,9 @@ package org.springframework.batch.repeat.interceptor; +import java.util.ArrayList; +import java.util.List; + import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.springframework.aop.ProxyMethodInvocation; @@ -26,13 +29,17 @@ import org.springframework.batch.repeat.RepeatException; import org.springframework.batch.repeat.RepeatOperations; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; /** - * A {@link MethodInterceptor} that can be used to automatically repeat calls to a method on a service. The injected - * {@link RepeatOperations} is used to control the completion of the loop. By default it will repeat until the target - * method returns null. Be careful when injecting a bespoke {@link RepeatOperations} that the loop will actually - * terminate, because the default policy for a vanilla {@link RepeatTemplate} will never complete if the return type of - * the target method is void (the value returned is always not-null, representing the {@link Void#TYPE}). + * A {@link MethodInterceptor} that can be used to automatically repeat calls to + * a method on a service. The injected {@link RepeatOperations} is used to + * control the completion of the loop. By default it will repeat until the + * target method returns null. Be careful when injecting a bespoke + * {@link RepeatOperations} that the loop will actually terminate, because the + * default policy for a vanilla {@link RepeatTemplate} will never complete if + * the return type of the target method is void (the value returned is always + * not-null, representing the {@link Void#TYPE}). * * @author Dave Syer */ @@ -52,44 +59,139 @@ public class RepeatOperationsInterceptor implements MethodInterceptor { } /** - * Invoke the proceeding method call repeatedly, according to the properties of the injected - * {@link RepeatOperations}. + * Invoke the proceeding method call repeatedly, according to the properties + * of the injected {@link RepeatOperations}. * * @see org.aopalliance.intercept.MethodInterceptor#invoke(org.aopalliance.intercept.MethodInvocation) */ public Object invoke(final MethodInvocation invocation) throws Throwable { - repeatOperations.iterate(new RepeatCallback() { + final List results = new ArrayList(); - public ExitStatus doInIteration(RepeatContext context) throws Exception { - try { + try { + repeatOperations.iterate(new RepeatCallback() { - MethodInvocation clone = invocation; - if (invocation instanceof ProxyMethodInvocation) { - clone = ((ProxyMethodInvocation) invocation).invocableClone(); - } else { - throw new IllegalStateException( - "MethodInvocation of the wrong type detected - this should not happen with Spring AOP, so please raise an issue if you see this exception"); + public ExitStatus doInIteration(RepeatContext context) throws Exception { + try { + + MethodInvocation clone = invocation; + if (invocation instanceof ProxyMethodInvocation) { + clone = ((ProxyMethodInvocation) invocation).invocableClone(); + } + else { + throw new IllegalStateException( + "MethodInvocation of the wrong type detected - this should not happen with Spring AOP, so please raise an issue if you see this exception"); + } + + // N.B. discards return value if there is one + if (clone.getMethod().getReturnType().equals(Void.TYPE)) { + clone.proceed(); + return ExitStatus.CONTINUABLE; + } + Object result = clone.proceed(); + if (!isComplete(result)) { + // We only save the last non-null result + results.clear(); + results.add(result); + return ExitStatus.CONTINUABLE; + } + else { + return ExitStatus.FINISHED; + } } - - // N.B. discards return value if there is one - if (clone.getMethod().getReturnType().equals(Void.TYPE)) { - clone.proceed(); - return ExitStatus.CONTINUABLE; - } - return new ExitStatus(clone.proceed() != null); - } catch (Throwable e) { - if (e instanceof Exception) { - throw (Exception) e; - } else { - throw new RepeatException("Unexpected error in batch interceptor", e); + catch (Throwable e) { + if (e instanceof Exception) { + throw (Exception) e; + } + else { + throw new RepeatOperationsInterceptorException("Unexpected error in batch interceptor", e); + } } } - } - }); + }); + } + catch (RepeatOperationsInterceptorException e) { + // Unwrap and re-throw any nasty errors + throw e.getCause(); + } + catch (Throwable t) { + throw t; + } - return null; + if (!results.isEmpty()) { + return results.get(0); + } + + Class returnType = invocation.getMethod().getReturnType(); + Object defaultValue = null; + if (ClassUtils.isPrimitiveOrWrapper(returnType)) { + defaultValue = getDefaultForPrimitiveType(returnType); + } + return defaultValue; + } + + /** + * @param result + * @return + */ + private boolean isComplete(Object result) { + return result == null || (result instanceof Boolean) && !((Boolean) result).booleanValue(); + } + + /** + * Simple wrapper exception class to enable nasty errors to be passed out of + * the scope of the repeat operations and handled by the caller. + * + * @author Dave Syer + * + */ + private static class RepeatOperationsInterceptorException extends RepeatException { + /** + * @param message + * @param e + */ + public RepeatOperationsInterceptorException(String message, Throwable e) { + super(message, e); + } + } + + /** + * Set up a default return value for primitive types (all basically "0"). + * @param returnType the desired primitive type + * @return a value to use as the default return value if recovery path is + * taken + */ + // TODO: cache these values. + private Object getDefaultForPrimitiveType(Class returnType) { + if (returnType.equals(Boolean.TYPE)) { + return Boolean.FALSE; + } + else if (returnType.equals(Byte.TYPE)) { + return Byte.valueOf("0"); + } + else if (returnType.equals(Character.TYPE)) { + return Character.valueOf('0'); + } + else if (returnType.equals(Short.TYPE)) { + return Short.valueOf("0"); + } + else if (returnType.equals(Integer.TYPE)) { + return Integer.valueOf('0'); + } + else if (returnType.equals(Long.TYPE)) { + return Long.valueOf('0'); + } + else if (returnType.equals(Float.TYPE)) { + return Float.valueOf('0'); + } + else if (returnType.equals(Double.TYPE)) { + return Double.valueOf('0'); + } + else if (returnType.equals(Void.TYPE)) { + return null; + } + throw new IllegalStateException("Primitive type with no default: " + returnType); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/RecoveryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/RecoveryCallback.java new file mode 100644 index 000000000..9157c12a0 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/RecoveryCallback.java @@ -0,0 +1,31 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.retry; + +/** + * @author Dave Syer + * + */ +public interface RecoveryCallback { + + /** + * @param throwable + * @return an Object that can be used to replace the callback result that + * failed + */ + Object recover(Throwable throwable); + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java deleted file mode 100644 index a64709d07..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java +++ /dev/null @@ -1,174 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.retry.callback; - -import org.springframework.batch.item.FailedItemIdentifier; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemRecoverer; -import org.springframework.batch.item.ItemWriter; -import org.springframework.batch.retry.RetryCallback; -import org.springframework.batch.retry.RetryContext; -import org.springframework.batch.retry.RetryException; -import org.springframework.batch.retry.RetryPolicy; -import org.springframework.batch.retry.policy.ItemWriterRetryPolicy; - -/** - * A {@link RetryCallback} that knows about and caches an item, and attempts to - * process it using an {@link ItemWriter}. Used by the - * {@link ItemWriterRetryPolicy} to enable external retry of the item - * processing. - * - * @author Dave Syer - * - * @see ItemWriterRetryPolicy - * @see RetryPolicy#handleRetryExhausted(RetryContext) - * - */ -public class ItemWriterRetryCallback implements RetryCallback { - - private Object item; - - private ItemWriter writer; - - private ItemRecoverer recoverer; - - private ItemKeyGenerator keyGenerator; - - private FailedItemIdentifier failedItemIdentifier; - - private ItemKeyGenerator defaultKeyGenerator = new ItemKeyGenerator() { - public Object getKey(Object item) { - return item; - } - }; - - /** - * Constructor with mandatory properties. - * - * @param item the item to process - * @param writer the writer to use to process it - */ - public ItemWriterRetryCallback(Object item, ItemWriter writer) { - super(); - this.item = item; - this.writer = writer; - } - - /** - * Setter for injecting optional recovery handler. If it is not injected but - * the reader or writer implement {@link ItemRecoverer}, one of those will - * be used instead (preferring the reader to the writer if both would be - * appropriate). - * - * @param recoverer - */ - public void setRecoverer(ItemRecoverer recoverer) { - this.recoverer = recoverer; - } - - /** - * Public setter for the {@link ItemKeyGenerator}. If it is not injected - * but the reader or writer implement {@link ItemKeyGenerator}, one of - * those will be used instead (preferring the reader to the writer if both - * would be appropriate). - * @param keyGenerator the keyGenerator to set - */ - public void setKeyGenerator(ItemKeyGenerator keyGenerator) { - this.keyGenerator = keyGenerator; - } - - /** - * Public setter for the {@link FailedItemIdentifier}. If it is not - * injected but the reader or writer implement {@link FailedItemIdentifier}, - * one of those will be used instead (preferring the reader to the writer if - * both would be appropriate). - * @param failedItemIdentifier the {@link FailedItemIdentifier} to set - */ - public void setFailedItemIdentifier(FailedItemIdentifier failedItemIdentifier) { - this.failedItemIdentifier = failedItemIdentifier; - } - - public Object doWithRetry(RetryContext context) throws Throwable { - // This requires a collaboration with the RetryPolicy... - if (!context.isExhaustedOnly()) { - if (item != null) { - writer.write(item); - } - return null; - } - throw new RetryException("Recovery path requested in retry callback."); - } - - public Object getItem() { - return item; - } - - /** - * Accessor for the {@link ItemRecoverer}. If the handler is null but the - * {@link ItemWriter} is an instance of {@link ItemRecoverer}, then it will - * be returned instead. If none of those strategies works then a default - * implementation of {@link ItemKeyGenerator} will be used that just returns - * the item. - * - * @return the {@link ItemRecoverer}. - */ - public ItemKeyGenerator getKeyGenerator() { - if (keyGenerator != null) { - return keyGenerator; - } - if (writer instanceof ItemKeyGenerator) { - return (ItemKeyGenerator) writer; - } - return defaultKeyGenerator; - } - - /** - * Accessor for the {@link FailedItemIdentifier}. If the handler is null - * but the {@link ItemWriter} or {@link ItemWriter} is an instance of - * {@link FailedItemIdentifier}, then it will be returned instead. If none - * of those strategies works returns null. - * - * @return the {@link FailedItemIdentifier}. - */ - public FailedItemIdentifier getFailedItemIdentifier() { - if (failedItemIdentifier != null) { - return failedItemIdentifier; - } - if (writer instanceof FailedItemIdentifier) { - return (FailedItemIdentifier) writer; - } - return null; - } - - /** - * Accessor for the {@link ItemRecoverer}. If the handler is null but the - * {@link ItemWriter} is an instance of {@link ItemRecoverer}, then it will - * be returned instead. - * - * @return the {@link ItemRecoverer}. - */ - public ItemRecoverer getRecoverer() { - if (recoverer != null) { - return recoverer; - } - if (writer instanceof ItemRecoverer) { - return (ItemRecoverer) writer; - } - return null; - } - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/RecoveryRetryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/RecoveryRetryCallback.java new file mode 100644 index 000000000..fac591fbc --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/RecoveryRetryCallback.java @@ -0,0 +1,139 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.retry.callback; + +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryException; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; + +/** + * A {@link RetryCallback} that knows about and caches an item, and attempts to + * process it using an {@link ItemWriter}. Used by the + * {@link RecoveryCallbackRetryPolicy} to enable external retry of the item + * processing. + * + * @author Dave Syer + * + * @see RecoveryCallbackRetryPolicy + * @see RetryPolicy#handleRetryExhausted(RetryContext) + * + */ +public class RecoveryRetryCallback implements RetryCallback { + + private final Object item; + + private final RetryCallback callback; + + private RecoveryCallback recoverer; + + private final Object key; + + private boolean forceRefresh = false; + + /** + * Constructor with mandatory properties. The key will be set to the item. + * + * @param item the item to process + * @param writer the writer to use to process it + */ + public RecoveryRetryCallback(Object item, RetryCallback writer) { + super(); + this.item = item; + this.callback = writer; + this.key = item; + } + + /** + * Constructor with mandatory properties. + * + * @param item the item to process + * @param writer the writer to use to process it + */ + public RecoveryRetryCallback(Object item, RetryCallback writer, Object key) { + super(); + this.item = item; + this.callback = writer; + this.key = key; + } + + /** + * Public getter for the key. This will be used to identify the item being + * processed, to see if it has previously failed. + * @return the key + */ + public Object getKey() { + return key; + } + + /** + * Setter for injecting optional recovery handler. If it is not injected but + * the reader or writer implement {@link ItemRecoverer}, one of those will + * be used instead (preferring the reader to the writer if both would be + * appropriate). + * + * @param recoverer + */ + public void setRecoveryCallback(RecoveryCallback recoverer) { + this.recoverer = recoverer; + } + + /** + * Public setter for a flag signalling to clients of this callback that the + * processing is not a retry. It is always safe to leave this set to the + * default value (false), but in some cases it is possible to determine by + * examining the input data whether a failure has never been encountered + * (e.g. a message header saying that the message has never been consumed). + * Clients who have this information can avoid a cache query in such cases + * by setting the flag to true. + * + * @param forceRefresh the flag value to set + */ + public void setForceRefresh(boolean forceRefresh) { + this.forceRefresh = forceRefresh; + } + + public Object doWithRetry(RetryContext context) throws Throwable { + if (!context.isExhaustedOnly()) { + return callback.doWithRetry(context); + } + // TODO: is this necessary? + throw new RetryException("Recovery path requested in retry callback."); + } + + public Object getItem() { + return item; + } + + public boolean isForceRefresh() { + return forceRefresh; + } + + /** + * Accessor for the {@link ItemRecoverer}. + * + * @return the {@link ItemRecoverer}. + */ + public RecoveryCallback getRecoveryCallback() { + return recoverer; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java new file mode 100644 index 000000000..c8f4946af --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java @@ -0,0 +1,141 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.retry.interceptor; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.item.ItemKeyGenerator; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.item.NewItemIdentifier; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.RetryPolicy; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; +import org.springframework.batch.retry.policy.NeverRetryPolicy; +import org.springframework.batch.retry.support.RetryTemplate; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; + +public class StatefulRetryOperationsInterceptor implements MethodInterceptor { + + private transient Log logger = LogFactory.getLog(getClass()); + + private ItemKeyGenerator keyGenerator; + + private ItemRecoverer recoverer; + + private NewItemIdentifier newItemIdentifier; + + private final RetryTemplate retryTemplate = new RetryTemplate(); + + /** + * + */ + public StatefulRetryOperationsInterceptor() { + super(); + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(new NeverRetryPolicy())); + } + + /** + * Public setter for the {@link ItemRecoverer} to use if the retry is + * exhausted. The recoverer should be able to return an object of the same + * type as the target object because its return value will be used to return + * to the caller in the case of a recovery. + * + * @param recoverer the {@link ItemRecoverer} to set + */ + public void setRecoverer(ItemRecoverer recoverer) { + this.recoverer = recoverer; + } + + public void setKeyGenerator(ItemKeyGenerator keyGenerator) { + this.keyGenerator = keyGenerator; + } + + /** + * Public setter for the retryPolicy. + * @param retryPolicy the retryPolicy to set + */ + public void setRetryPolicy(RetryPolicy retryPolicy) { + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(retryPolicy)); + } + + /** + * Public setter for the {@link NewItemIdentifier}. Only set this if the + * arguments to the intercepted method can be inspected to find out if they + * have never been processed before. + * @param newItemIdentifier the {@link NewItemIdentifier} to set + */ + public void setNewItemIdentifier(NewItemIdentifier newItemIdentifier) { + this.newItemIdentifier = newItemIdentifier; + } + + /** + * Wrap the method invocation in a stateful retry with the policy and other + * helpers provided. If there is a failure the exception will generally be + * re-thrown. The only time it is not re-thrown is when retry is exhausted + * and the recovery path is taken (though the {@link ItemRecoverer} provided + * if there is one). In that case the value returned from the method + * invocation will be null, or if primitive then "0" (e.g. Boolean.FALSE, 0L + * etc.). + * + * @see org.aopalliance.intercept.MethodInterceptor#invoke(org.aopalliance.intercept.MethodInvocation) + */ + public Object invoke(final MethodInvocation invocation) throws Throwable { + + logger.debug("Executing proxied method in stateful retry: " + invocation.getStaticPart() + "(" + + ObjectUtils.getIdentityHexString(invocation) + ")"); + + Object[] args = invocation.getArguments(); + Assert.state(args.length > 0, "Stateful retry applied to method that takes no arguments: " + + invocation.getStaticPart()); + Object arg = args; + if (args.length == 1) { + arg = args[0]; + } + final Object item = arg; + + RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + return invocation.proceed(); + } + }, keyGenerator != null ? keyGenerator.getKey(item) : item); + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable cause) { + if (recoverer != null) { + return recoverer.recover(item, cause); + } + return item; + } + }); + if (newItemIdentifier != null) { + callback.setForceRefresh(newItemIdentifier.isNew(item)); + } + + Object result = retryTemplate.execute(callback); + + logger.debug("Exiting proxied method in stateful retry with result: (" + result + ")"); + + return result; + + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java similarity index 71% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java index 2f11d3335..e9a3ed07b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java @@ -18,34 +18,32 @@ package org.springframework.batch.retry.policy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.batch.item.FailedItemIdentifier; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.repeat.support.RepeatSynchronizationManager; +import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; import org.springframework.batch.retry.RetryException; import org.springframework.batch.retry.RetryPolicy; import org.springframework.batch.retry.TerminatedRetryException; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; import org.springframework.batch.retry.context.RetryContextSupport; import org.springframework.util.Assert; /** - * A {@link RetryPolicy} that detects an {@link ItemWriterRetryCallback} when it + * A {@link RetryPolicy} that detects an {@link RecoveryRetryCallback} when it * opens a new context, and uses it to make sure the item is in place for later * decisions about how to retry or backoff. The callback should be an instance - * of {@link ItemWriterRetryCallback} otherwise an exception will be thrown when + * of {@link RecoveryRetryCallback} otherwise an exception will be thrown when * the context is created. * * @author Dave Syer * */ -public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { +public class RecoveryCallbackRetryPolicy extends AbstractStatefulRetryPolicy { protected Log logger = LogFactory.getLog(getClass()); - public static final String EXHAUSTED = ItemWriterRetryPolicy.class.getName() + ".EXHAUSTED"; + public static final String EXHAUSTED = RecoveryCallbackRetryPolicy.class.getName() + ".EXHAUSTED"; private RetryPolicy delegate; @@ -54,7 +52,7 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { * * @param delegate */ - public ItemWriterRetryPolicy(RetryPolicy delegate) { + public RecoveryCallbackRetryPolicy(RetryPolicy delegate) { super(); this.delegate = delegate; } @@ -63,7 +61,7 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { * Default constructor. Creates a new {@link SimpleRetryPolicy} for the * delegate. */ - public ItemWriterRetryPolicy() { + public RecoveryCallbackRetryPolicy() { this(new SimpleRetryPolicy()); } @@ -97,7 +95,7 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { /** * Create a new context for the execution of the callback, which must be an - * instance of {@link ItemWriterRetryCallback}. + * instance of {@link RecoveryRetryCallback}. * * @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback, * RetryContext) @@ -106,8 +104,8 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { * type. */ public RetryContext open(RetryCallback callback, RetryContext parent) { - Assert.state(callback instanceof ItemWriterRetryCallback, "Callback must be ItemProviderRetryCallback"); - ItemWriterRetryContext context = new ItemWriterRetryContext((ItemWriterRetryCallback) callback, parent); + Assert.state(callback instanceof RecoveryRetryCallback, "Callback must be ItemProviderRetryCallback"); + ItemWriterRetryContext context = new ItemWriterRetryContext((RecoveryRetryCallback) callback, parent); context.open(callback, null); return context; } @@ -135,8 +133,6 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { private class ItemWriterRetryContext extends RetryContextSupport implements RetryPolicy { - final private Object item; - final private Object key; final private int initialHashCode; @@ -144,19 +140,15 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { // The delegate context... private RetryContext delegateContext; - final private ItemRecoverer recoverer; + final private RecoveryCallback recoverer; - final private ItemKeyGenerator keyGenerator; + final private boolean forceRefresh; - final private FailedItemIdentifier failedItemIdentifier; - - public ItemWriterRetryContext(ItemWriterRetryCallback callback, RetryContext parent) { + public ItemWriterRetryContext(RecoveryRetryCallback callback, RetryContext parent) { super(parent); - this.recoverer = callback.getRecoverer(); - this.keyGenerator = callback.getKeyGenerator(); - this.item = callback.getItem(); - this.key = keyGenerator.getKey(item); - this.failedItemIdentifier = callback.getFailedItemIdentifier(); + this.recoverer = callback.getRecoveryCallback(); + this.key = callback.getKey(); + this.forceRefresh = callback.isForceRefresh(); this.initialHashCode = key.hashCode(); } @@ -169,7 +161,11 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { } public RetryContext open(RetryCallback callback, RetryContext parent) { - if (hasFailed(failedItemIdentifier, key)) { + if (forceRefresh) { + // Avoid a cache hit if the caller tells us this is a fresh item + this.delegateContext = delegate.open(callback, null); + } + else if (retryContextCache.containsKey(key)) { this.delegateContext = retryContextCache.get(key); if (this.delegateContext == null) { throw new RetryException("Inconsistent state for failed item: no history found. " @@ -213,14 +209,9 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { retryContextCache.remove(key); RepeatSynchronizationManager.setCompleteOnly(); if (recoverer != null) { - boolean success = recoverer.recover(item, context.getLastThrowable()); - if (!success) { - int count = context.getRetryCount(); - logger.error("Could not recover from error after retry exhausted after [" + count + "] attempts.", - context.getLastThrowable()); - } + return recoverer.recover(context.getLastThrowable()); } - return item; + return null; } public Throwable getLastThrowable() { @@ -233,23 +224,4 @@ public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy { } - /** - * Extension point for cases where it is possible to avoid a cache hit by - * inspecting the item to determine if could ever have been seen before. In - * a messaging environment where the item is a message, it can be inspected - * to see if it has been delivered before.
- * - * The default implementation of this method checks for a non-null - * {@link FailedItemIdentifier}. Otherwise we just check the cache for the - * item key. - * - * @param failedItemIdentifier - * @param key - */ - protected boolean hasFailed(FailedItemIdentifier failedItemIdentifier, Object key) { - if (failedItemIdentifier != null) { - return failedItemIdentifier.hasFailed(key); - } - return retryContextCache.containsKey(key); - } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java index 65987d059..0819f90b6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java @@ -21,8 +21,8 @@ import junit.framework.TestCase; public class ItemRecoveryHandlerTests extends TestCase { ItemRecoverer recoverer = new ItemRecoverer() { - public boolean recover(Object data, Throwable cause) { - return false; + public Object recover(Object data, Throwable cause) { + return null; } }; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/jms/JmsItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/jms/JmsItemReaderTests.java index cef265d6a..e0677a9f3 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/jms/JmsItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/jms/JmsItemReaderTests.java @@ -171,12 +171,12 @@ public class JmsItemReaderTests extends TestCase { messageControl.replay(); itemProvider.setItemType(Message.class); - assertEquals(true, itemProvider.hasFailed(message)); + assertEquals(false, itemProvider.isNew(message)); messageControl.verify(); } public void testIsNewForNonMessage() throws Exception { itemProvider.setItemType(String.class); - assertEquals(true, itemProvider.hasFailed("foo")); + assertEquals(false, itemProvider.isNew("foo")); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/StubItemKeyGeneratorRecoverer.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/StubItemKeyGeneratorRecoverer.java index d39160d51..e6051e3f6 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/StubItemKeyGeneratorRecoverer.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/StubItemKeyGeneratorRecoverer.java @@ -22,19 +22,19 @@ import org.springframework.batch.item.ItemRecoverer; public class StubItemKeyGeneratorRecoverer implements ItemRecoverer, ItemKeyGenerator { /** - * Do nothing. Subclassses should override to implement recovery behaviour. + * Do nothing and return null. Subclassses should override to implement + * recovery behaviour. * * @see org.springframework.batch.item.ItemRecoverer#recover(java.lang.Object, * Throwable) * - * @return false if nothing can be done (the default), or true if the item - * can now safely be ignored or committed. + * @return null. */ - public boolean recover(Object item, Throwable cause) { - return false; + public Object recover(Object item, Throwable cause) { + return null; } - - /** + + /** * Return the item (assume it is its own key). * * @see org.springframework.batch.item.ItemKeyGenerator#getKey(java.lang.Object) diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/RecoveryRetryCallbackTests.java similarity index 79% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/RecoveryRetryCallbackTests.java index 9d4c2bb30..24cbda858 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/RecoveryRetryCallbackTests.java @@ -21,16 +21,16 @@ import java.util.List; import junit.framework.TestCase; -import org.springframework.batch.item.AbstractItemWriter; -import org.springframework.batch.retry.StubItemKeyGeneratorRecoverer; +import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; import org.springframework.batch.retry.RetryException; +import org.springframework.batch.retry.StubItemKeyGeneratorRecoverer; import org.springframework.batch.retry.TerminatedRetryException; import org.springframework.batch.retry.context.RetryContextSupport; import org.springframework.batch.retry.policy.NeverRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; -public class ItemWriterRetryCallbackTests extends TestCase { +public class RecoveryRetryCallbackTests extends TestCase { List calls = new ArrayList(); @@ -40,32 +40,28 @@ public class ItemWriterRetryCallbackTests extends TestCase { StubItemKeyGeneratorRecoverer recoverer; - ItemWriterRetryCallback callback; - - private AbstractItemWriter writer; + RecoveryRetryCallback callback; protected void setUp() throws Exception { super.setUp(); template = new RetryTemplate(); recoverer = new StubItemKeyGeneratorRecoverer() { - public boolean recover(Object data, Throwable cause) { + public Object recover(Object data, Throwable cause) { count++; calls.add(data); - return true; + return data; } + public Object getKey(Object item) { return "key" + (count++); } }; - writer = new AbstractItemWriter() { - public void write(Object data) { + callback = new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - if (data.equals("bar")) { - throw new IllegalStateException("Bar detected"); - } + return null; } - }; - callback = new ItemWriterRetryCallback("foo", writer); + }); } public void testDoWithRetrySuccessfulFirstTime() throws Exception { @@ -76,12 +72,17 @@ public class ItemWriterRetryCallbackTests extends TestCase { public void testContextInitializedWithItemAndCanRetry() throws Exception { // We can use the policy to intercept the context and do something with // the item... - callback = new ItemWriterRetryCallback("bar", writer); + callback = new RecoveryRetryCallback("bar", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + count++; + throw new IllegalStateException("Detected bar"); + } + }); assertEquals(0, calls.size()); template.setRetryPolicy(new NeverRetryPolicy() { public boolean canRetry(RetryContext context) { // ...register the failed item - calls.add("item(" + count + ")=" + callback.getItem()); + calls.add("item(" + count + ")=" + callback.getKey()); // Do not call the base class method - the attempt counts as // successful now if (count < 2) // only retry once @@ -107,12 +108,17 @@ public class ItemWriterRetryCallbackTests extends TestCase { public void testContextInitializedWithItemAndRegisterThrowable() throws Exception { // We can use the policy to intercept the context and do something with // the item... - callback = new ItemWriterRetryCallback("bar", writer); + callback = new RecoveryRetryCallback("bar", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + count++; + throw new IllegalStateException("Detected bar"); + } + }); assertEquals(0, calls.size()); template.setRetryPolicy(new NeverRetryPolicy() { public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException { // ...register the failed item - calls.add("item=" + callback.getItem()); + calls.add("item=" + callback.getKey()); // Call the base class method so that the next attempt is a // failure. super.registerThrowable(context, throwable); @@ -143,8 +149,8 @@ public class ItemWriterRetryCallbackTests extends TestCase { } public void testGetKey() throws Exception { - callback.setKeyGenerator(recoverer); - assertEquals("key0", callback.getKeyGenerator().getKey("foo")); + callback = new RecoveryRetryCallback("foo", null, "key0"); + assertEquals("key0", callback.getKey()); } public void testRecoverWithoutSession() throws Exception { diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java new file mode 100644 index 000000000..7fe6b3f32 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java @@ -0,0 +1,212 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.retry.interceptor; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +import junit.framework.TestCase; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.target.SingletonTargetSource; +import org.springframework.batch.item.ItemRecoverer; +import org.springframework.batch.retry.policy.AlwaysRetryPolicy; +import org.springframework.batch.retry.policy.NeverRetryPolicy; +import org.springframework.batch.retry.policy.SimpleRetryPolicy; + +/** + * @author Dave Syer + * + */ +public class StatefulRetryOperationsInterceptorTests extends TestCase { + + private StatefulRetryOperationsInterceptor interceptor; + + private Service service; + + private Transformer transformer; + + private static int count; + + public void setUp() throws Exception { + interceptor = new StatefulRetryOperationsInterceptor(); + service = (Service) ProxyFactory.getProxy(Service.class, new SingletonTargetSource(new ServiceImpl())); + transformer = (Transformer) ProxyFactory.getProxy(Transformer.class, new SingletonTargetSource(new TransformerImpl())); + count = 0; + } + + public void testDefaultInterceptorSunnyDay() throws Exception { + ((Advised) service).addAdvice(interceptor); + try { + service.service("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + } + + public void testDefaultTransformerInterceptorSunnyDay() throws Exception { + ((Advised) transformer).addAdvice(interceptor); + try { + transformer.transform("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + } + + public void testDefaultInterceptorAlwaysRetry() throws Exception { + interceptor.setRetryPolicy(new AlwaysRetryPolicy()); + ((Advised) service).addAdvice(interceptor); + try { + service.service("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + } + + public void testInterceptorChainWithRetry() throws Exception { + ((Advised) service).addAdvice(interceptor); + final List list = new ArrayList(); + ((Advised) service).addAdvice(new MethodInterceptor() { + public Object invoke(MethodInvocation invocation) throws Throwable { + list.add("chain"); + return invocation.proceed(); + } + }); + interceptor.setRetryPolicy(new SimpleRetryPolicy(2)); + try { + service.service("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + service.service("foo"); + assertEquals(2, count); + assertEquals(2, list.size()); + } + + public void testTransformerWithSuccessfulRetry() throws Exception { + ((Advised) transformer).addAdvice(interceptor); + interceptor.setRetryPolicy(new SimpleRetryPolicy(2)); + try { + transformer.transform("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + Collection result = transformer.transform("foo"); + assertEquals(2, count); + assertEquals(1, result.size()); + } + + public void testRetryExceptionAfterTooManyAttempts() throws Exception { + ((Advised) service).addAdvice(interceptor); + interceptor.setRetryPolicy(new NeverRetryPolicy()); + try { + service.service("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + interceptor.setRecoverer(new ItemRecoverer() { + public Object recover(Object data, Throwable cause) { + count++; + return data; + } + }); + service.service("foo"); + assertEquals(2, count); + } + + public void testTransformerRecoveryAfterTooManyAttempts() throws Exception { + ((Advised) transformer).addAdvice(interceptor); + interceptor.setRetryPolicy(new NeverRetryPolicy()); + try { + transformer.transform("foo"); + fail("Expected Exception."); + } + catch (Exception e) { + String message = e.getMessage(); + assertTrue("Wrong message: " + message, message.startsWith("Not enough calls")); + } + assertEquals(1, count); + interceptor.setRecoverer(new ItemRecoverer() { + public Object recover(Object data, Throwable cause) { + count++; + return Collections.singleton(data); + } + }); + Collection result = transformer.transform("foo"); + assertEquals(2, count); + assertEquals(1, result.size()); + } + + public static interface Service { + void service(String in) throws Exception; + } + + public static class ServiceImpl implements Service { + + public void service(String in) throws Exception { + count++; + if (count < 2) { + throw new Exception("Not enough calls: " + count); + } + } + + } + + public static interface Transformer { + Collection transform(String in) throws Exception; + } + + public static class TransformerImpl implements Transformer { + + public Collection transform(String in) throws Exception { + count++; + if (count < 2) { + throw new Exception("Not enough calls: " + count); + } + return Collections.singleton(in + ":" + count); + } + + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/RecoveryRetryPolicyTests.java similarity index 63% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/RecoveryRetryPolicyTests.java index 6e09a0518..58585158c 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/RecoveryRetryPolicyTests.java @@ -17,56 +17,40 @@ package org.springframework.batch.retry.policy; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import junit.framework.TestCase; -import org.springframework.batch.item.AbstractItemWriter; -import org.springframework.batch.item.FailedItemIdentifier; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.support.ListItemReader; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.context.RepeatContextSupport; import org.springframework.batch.repeat.support.RepeatSynchronizationManager; +import org.springframework.batch.retry.RecoveryCallback; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; import org.springframework.batch.retry.RetryException; -import org.springframework.batch.retry.StubItemKeyGeneratorRecoverer; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; -import org.springframework.batch.retry.context.RetryContextSupport; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; import org.springframework.batch.retry.support.RetryTemplate; -public class ItemWriterRetryPolicyTests extends TestCase { +public class RecoveryRetryPolicyTests extends TestCase { - private ItemWriterRetryPolicy policy = new ItemWriterRetryPolicy(); - - private StubItemKeyGeneratorRecoverer recoverer; + private RecoveryCallbackRetryPolicy policy = new RecoveryCallbackRetryPolicy(); private int count = 0; private List list = new ArrayList(); - protected void setUp() throws Exception { - super.setUp(); - // The list simulates a failed delivery, redelivery of the same message, - // then a new message... - recoverer = new StubItemKeyGeneratorRecoverer() { - public boolean recover(Object data, Throwable cause) { + public void testOpenSunnyDay() throws Exception { + + final StringHolder item = new StringHolder("foo"); + RetryCallback writer = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); - return true; + list.add(item.string); + return item; } }; - } - public void testOpenSunnyDay() throws Exception { - RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { - count++; - list.add(data); - } - }), null); + RetryContext context = policy.open(new RecoveryRetryCallback("foo", writer), null); assertNotNull(context); // we haven't called the processor yet... assertEquals(0, count); @@ -89,9 +73,10 @@ public class ItemWriterRetryPolicyTests extends TestCase { public void testCanRetry() { policy.setDelegate(new AlwaysRetryPolicy()); - RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RetryContext context = policy.open(new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; + return null; } }), null); assertNotNull(context); @@ -102,10 +87,10 @@ public class ItemWriterRetryPolicyTests extends TestCase { public void testRegisterThrowable() { policy.setDelegate(new NeverRetryPolicy()); - RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RetryContext context = policy.open(new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }), null); assertNotNull(context); @@ -115,10 +100,10 @@ public class ItemWriterRetryPolicyTests extends TestCase { public void testClose() throws Exception { policy.setDelegate(new NeverRetryPolicy()); - RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RetryContext context = policy.open(new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }), null); assertNotNull(context); @@ -131,10 +116,10 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testOpenTwice() throws Exception { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RecoveryRetryCallback callback = new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }); policy.setDelegate(new SimpleRetryPolicy(2)); @@ -156,13 +141,21 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testRecover() throws Exception { - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + final String input = "foo"; + RecoveryRetryCallback callback = new RecoveryRetryCallback(input, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + return null; + } + }); + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable cause) { + count++; + list.add(input); + return input; } }); - callback.setRecoverer(recoverer); RetryContext context = policy.open(callback, null); assertNotNull(context); assertTrue(policy.canRetry(context)); @@ -172,7 +165,7 @@ public class ItemWriterRetryPolicyTests extends TestCase { context = policy.open(callback, null); // On the second retry, the recovery path is taken... Object result = policy.handleRetryExhausted(context); - assertNotNull(result); // default result is null + assertEquals("foo", result); // the recoverer returns the item assertEquals(1, count); assertFalse(policy.canRetry(context)); assertEquals("foo", list.get(0)); @@ -186,25 +179,22 @@ public class ItemWriterRetryPolicyTests extends TestCase { RepeatSynchronizationManager.clear(); } - public void testFailedItemIdentifier() throws Exception { - policy = new ItemWriterRetryPolicy(); - policy.setDelegate(new SimpleRetryPolicy(1)); - MockFailedItemProvider provider = new MockFailedItemProvider(Collections.EMPTY_LIST); - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", null); - callback.setFailedItemIdentifier(provider); - policy.open(callback, null); - assertEquals(1, provider.hasFailedCount); - } - public void testRecoverWithTemplate() throws Exception { - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + final String input = "foo"; + RecoveryRetryCallback callback = new RecoveryRetryCallback(input, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { throw new RuntimeException("Barf!"); } }); - callback.setRecoverer(recoverer); + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable cause) { + count++; + list.add(input); + return input; + } + }); RetryTemplate template = new RetryTemplate(); template.setRetryPolicy(policy); Object result = null; @@ -217,16 +207,16 @@ public class ItemWriterRetryPolicyTests extends TestCase { } // On the second retry, the recovery path is taken... result = template.execute(callback); - assertNotNull(result); // default result is last item processed + assertEquals(input, result); // default result is the item assertEquals(1, count); - assertEquals("foo", list.get(0)); + assertEquals(input, list.get(0)); } public void testExhaustedClearsHistoryAfterLastAttempt() throws Exception { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RecoveryRetryCallback callback = new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }); policy.setDelegate(new SimpleRetryPolicy(1)); @@ -241,7 +231,7 @@ public class ItemWriterRetryPolicyTests extends TestCase { assertFalse(policy.canRetry(context)); policy.close(context); Object result = policy.handleRetryExhausted(context); - assertEquals("foo", result); // default result is last item + assertNull(result); // default result is null context = policy.open(callback, null); // True after exhausted - the history is reset... @@ -249,12 +239,12 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testRetryCount() throws Exception { - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); - RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() { - public void write(Object data) { + RetryContext context = policy.open(new RecoveryRetryCallback("foo", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }), null); assertNotNull(context); @@ -266,14 +256,14 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testRetryCountPreservedBetweenRetries() throws Exception { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback("bar", new AbstractItemWriter() { - public void write(Object data) { + RecoveryRetryCallback callback = new RecoveryRetryCallback("bar", new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + return null; } }); - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); RetryContext context = policy.open(callback, null); assertNotNull(context); @@ -285,29 +275,23 @@ public class ItemWriterRetryPolicyTests extends TestCase { assertEquals(2, context.getRetryCount()); } - public void testSetCacheAndHasFailed() throws Exception { - MapRetryContextCache cache = new MapRetryContextCache(); - policy.setRetryContextCache(cache); - cache.put("foo", new RetryContextSupport(null)); - assertTrue(policy.hasFailed(null, "foo")); - } - public void testKeyGeneratorNotConsistentAfterFailure() throws Throwable { - AbstractItemWriter writer = new AbstractItemWriter() { - public void write(Object data) { + policy = new RecoveryCallbackRetryPolicy(); + policy.setDelegate(new SimpleRetryPolicy(3)); + final StringHolder item = new StringHolder("bar"); + + RetryCallback writer = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { // This simulates what happens if someone uses a primary key // for hasCode and equals and then relies on default key // generator - ((StringHolder) data).string = ((StringHolder) data).string + (count++); + ((StringHolder) item).string = ((StringHolder) item).string + (count++); throw new RuntimeException("Barf!"); } }; - policy = new ItemWriterRetryPolicy(); - policy.setDelegate(new SimpleRetryPolicy(3)); - StringHolder item = new StringHolder("bar"); - ItemWriterRetryCallback callback = new ItemWriterRetryCallback(item, writer); + RecoveryRetryCallback callback = new RecoveryRetryCallback(item, writer); RetryContext context = policy.open(callback, null); assertNotNull(context); try { @@ -330,20 +314,24 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testCacheCapacity() throws Exception { - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); policy.setRetryContextCache(new MapRetryContextCache(1)); - AbstractItemWriter writer = new AbstractItemWriter() { - public void write(Object data) { + final StringHolder item = new StringHolder("foo"); + + RetryCallback writer = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + list.add(item.string); + return item; } }; RetryContext context; - context = policy.open(new ItemWriterRetryCallback("foo", writer), null); + context = policy.open(new RecoveryRetryCallback(item, writer), null); policy.registerThrowable(context, null); assertEquals(0, context.getRetryCount()); - context = policy.open(new ItemWriterRetryCallback("bar", writer), null); + item.string = "bar"; + context = policy.open(new RecoveryRetryCallback(item, writer), null); try { policy.registerThrowable(context, new RuntimeException("foo")); fail("Expected RetryException"); @@ -355,48 +343,32 @@ public class ItemWriterRetryPolicyTests extends TestCase { } public void testCacheCapacityNotReachedIfRecovered() throws Exception { - policy = new ItemWriterRetryPolicy(); + policy = new RecoveryCallbackRetryPolicy(); policy.setDelegate(new SimpleRetryPolicy(1)); policy.setRetryContextCache(new MapRetryContextCache(2)); - AbstractItemWriter writer = new AbstractItemWriter() { - public void write(Object data) { + final StringHolder item = new StringHolder("foo"); + + RetryCallback writer = new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { count++; - list.add(data); + list.add(item.string); + return item; } }; + RetryContext context; - context = policy.open(new ItemWriterRetryCallback("foo", writer), null); + context = policy.open(new RecoveryRetryCallback(item, writer), null); policy.registerThrowable(context, null); assertEquals(0, context.getRetryCount()); policy.registerThrowable(context, new RuntimeException("foo")); - context = policy.open(new ItemWriterRetryCallback("bar", writer), null); + context = policy.open(new RecoveryRetryCallback("bar", writer), null); policy.registerThrowable(context, null); policy.handleRetryExhausted(context); - context = policy.open(new ItemWriterRetryCallback("spam", writer), null); + context = policy.open(new RecoveryRetryCallback("spam", writer), null); policy.registerThrowable(context, null); assertEquals(0, context.getRetryCount()); } - private static class MockFailedItemProvider extends ListItemReader implements ItemKeyGenerator, - FailedItemIdentifier { - - private int hasFailedCount = 0; - - public MockFailedItemProvider(List list) { - super(list); - } - - public boolean hasFailed(Object item) { - hasFailedCount++; - return false; - } - - public Object getKey(Object item) { - throw new UnsupportedOperationException("Should not call this method"); - } - - } - private static class StringHolder { private String string; diff --git a/spring-batch-integration/.springBeans b/spring-batch-integration/.springBeans index a17ff62ef..d32d899ee 100644 --- a/spring-batch-integration/.springBeans +++ b/spring-batch-integration/.springBeans @@ -1,8 +1,11 @@ - - xml - + 1 + + + + + src/test/resources/org/springframework/batch/jms/jms-context.xml src/test/resources/data-source.xml diff --git a/spring-batch-integration/pom.xml b/spring-batch-integration/pom.xml index a98084a5c..4f86d2fcf 100644 --- a/spring-batch-integration/pom.xml +++ b/spring-batch-integration/pom.xml @@ -200,7 +200,6 @@ org.springframework spring-aop - test diff --git a/spring-batch-integration/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java b/spring-batch-integration/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java new file mode 100644 index 000000000..1a5d936ca --- /dev/null +++ b/spring-batch-integration/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java @@ -0,0 +1,138 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.container.jms; + +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.Session; + +import org.aopalliance.aop.Advice; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.DefaultPointcutAdvisor; +import org.springframework.aop.support.NameMatchMethodPointcut; +import org.springframework.batch.repeat.RepeatOperations; +import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; +import org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.transaction.interceptor.TransactionInterceptor; + +/** + * Message listener container adapted for intercepting the message reception + * with advice provided through configuration.
+ * + * To enable batching of messages in a single transaction, use the + * {@link TransactionInterceptor} and the {@link RepeatOperationsInterceptor} in + * the advice chain (with or without a transaction manager set in the base + * class). Instead of receiving a single message and processing it, the + * container will then use a {@link RepeatOperations} to receive multiple + * messages in the same thread. Use with a {@link RepeatOperations} and a + * transaction interceptor. If the transaction interceptor uses XA then use an + * XA connection factory, or else the + * {@link TransactionAwareConnectionFactoryProxy} to synchronize the JMS session + * with the ongoing transaction (opening up the possibility of duplicate + * messages after a failure). In the latter case you will not need to provide a + * transaction manager in the base class - it only gets on the way and prevents + * the JMS session from synchronizing with the database transaction. + * + * @author Dave Syer + * + */ +public class BatchMessageListenerContainer extends DefaultMessageListenerContainer { + + /** + * @author Dave Syer + * + */ + public static interface ContainerDelegate { + boolean receiveAndExecute(Session session, MessageConsumer consumer) throws JMSException; + } + + private Advice[] advices = new Advice[0]; + + private ContainerDelegate delegate = new ContainerDelegate() { + public boolean receiveAndExecute(Session session, MessageConsumer consumer) throws JMSException { + return BatchMessageListenerContainer.super.receiveAndExecute(session, consumer); + } + }; + + private ContainerDelegate proxy = delegate; + + /** + * Public setter for the {@link Advice}. + * @param advices the advice to set + */ + public void setAdvices(Advice[] advices) { + this.advices = advices; + } + + /** + * Set up interceptor with provided advice on the + * {@link #receiveAndExecute(Session, MessageConsumer)} method. + * + * @see org.springframework.jms.listener.AbstractJmsListeningContainer#afterPropertiesSet() + */ + public void afterPropertiesSet() { + super.afterPropertiesSet(); + ProxyFactory factory = new ProxyFactory(); + for (int i = 0; i < advices.length; i++) { + DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advices[i]); + NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut(); + pointcut.addMethodName("receiveAndExecute"); + advisor.setPointcut(pointcut); + factory.addAdvisor(advisor); + } + factory.setProxyTargetClass(false); + factory.addInterface(ContainerDelegate.class); + factory.setTarget(delegate); + proxy = (ContainerDelegate) factory.getProxy(); + } + + /** + * Override base class to prevent exceptions from being swallowed. Should be + * an injectable strategy (see SPR-4733). + * + * @see org.springframework.jms.listener.AbstractMessageListenerContainer#handleListenerException(java.lang.Throwable) + */ + protected void handleListenerException(Throwable ex) { + if (!isSessionTransacted()) { + // Log the exceptions in base class if not transactional anyway + super.handleListenerException(ex); + return; + } + logger.debug("Re-throwing exception in container."); + if (ex instanceof RuntimeException) { + // We need to re-throw so that an enclosing non-JMS transaction can + // rollback... + throw (RuntimeException) ex; + } + else if (ex instanceof Error) { + // Just re-throw Error instances because otherwise unit tests just + // swallow exceptions from EasyMock and JUnit. + throw (Error) ex; + } + } + + /** + * Override base class method to wrap call in advice if provided. + * @see org.springframework.jms.listener.AbstractPollingMessageListenerContainer#receiveAndExecute(javax.jms.Session, + * javax.jms.MessageConsumer) + */ + protected boolean receiveAndExecute(final Session session, final MessageConsumer consumer) throws JMSException { + return proxy.receiveAndExecute(session, consumer); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java b/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java deleted file mode 100644 index 1aef897a2..000000000 --- a/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java +++ /dev/null @@ -1,137 +0,0 @@ -/* - * Copyright 2006-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.batch.container.jms; - -import javax.jms.JMSException; -import javax.jms.MessageConsumer; -import javax.jms.Session; - -import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy; -import org.springframework.jms.listener.DefaultMessageListenerContainer; -import org.springframework.jms.listener.adapter.MessageListenerAdapter; - -/** - * Message listener container adapted for batching the message processing. - * Instead of receiving a single message and processing it, we use a - * {@link RepeatOperations} to receive multiple messages in the same thread. Use - * with a transactional {@link RepeatOperations} and either an XA connection - * factory, or the {@link TransactionAwareConnectionFactoryProxy} to synchronize - * the JMS session with an ongoing transaction. - * - * @author Dave Syer - * - */ -public class BatchMessageListenerContainer extends DefaultMessageListenerContainer { - - private RepeatOperations template; - - /** - * Create a new {@link BatchMessageListenerContainer}. The container is set - * with auto startup = false (not the default of the parent container). - * - * @param template a {@link RepeatOperations}. It is advisable to set the - * {@link RepeatOperations} with a sensible termination policy, like a small - * fixed chunk size. - */ - public BatchMessageListenerContainer(RepeatOperations template) { - super(); - this.template = template; - setAutoStartup(false); - // Avoid error on startup... - // http://opensource.atlassian.com/projects/spring/browse/SPR-3154 - setMessageListener(new MessageListenerAdapter()); - } - - /** - * Override base class to prevent exceptions from being swallowed. - * - * @see org.springframework.jms.listener.AbstractMessageListenerContainer#handleListenerException(java.lang.Throwable) - */ - protected void handleListenerException(Throwable ex) { - if (!isSessionTransacted()) { - // Log the exceptions in base class if not transactional anyway - super.handleListenerException(ex); - return; - } - logger.debug("Re-throwing exception in container."); - if (ex instanceof RuntimeException) { - // We need to re-throw so that an enclosing non-JMS transaction can - // rollback... - throw (RuntimeException) ex; - } - else if (ex instanceof Error) { - // Just re-throw Error instances because otherwise unit tests just - // swallow exceptions from EasyMock and JUnit. - throw (Error) ex; - } - } - - /** - * Override base class method to wrap call in a batch. - * @see org.springframework.jms.listener.AbstractPollingMessageListenerContainer#receiveAndExecute(javax.jms.Session, - * javax.jms.MessageConsumer) - */ - protected boolean receiveAndExecute(final Session session, final MessageConsumer consumer) throws JMSException { - - ExitStatus status = template.iterate(new RepeatCallback() { - public ExitStatus doInIteration(RepeatContext context) throws Exception { - return doBatchCallBack(session, consumer); - } - }); - - return status.isContinuable(); - } - - /** - * Wraps a call to {@link #receiveAndExecute(Session, MessageConsumer)}, - * retrieving the message from thread local. - * - * @param session - * @param consumer - * - * @throws JMSException - * - * @see #receiveMessage(MessageConsumer) - */ - protected ExitStatus doBatchCallBack(Session session, MessageConsumer consumer) throws JMSException { - /* - * The base class receiveAndExecute is transactional (if configured). We - * could extend the tx boundary to the whole batch by making the - * template.execute transactional, and either switch off the tx manager - * in this object, or live with its default propagation=REQUIRED - * behaviour. - * - * But if the super class transaction manager is a - * JmsTransactionManager, which is the normal choice for a message - * listener container (see - * http://opensource.atlassian.com/projects/spring/browse/SPR-3156), - * then it will not behave as expected. In particular since the - * JmsTransactionManager is not aware of the batch template (execute or - * callback) transactions, it will commit message sessions that should - * be rolled back when a batch fails. - */ - if (super.receiveAndExecute(session, consumer)) { - return ExitStatus.CONTINUABLE; - } - return ExitStatus.FINISHED; - } - -} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java index 1b3f4c883..5fc5dd32a 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java @@ -15,18 +15,16 @@ */ package org.springframework.batch.container.jms; -import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; -import javax.jms.TextMessage; -import org.springframework.batch.item.AbstractItemWriter; -import org.springframework.batch.item.ItemKeyGenerator; -import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.jms.ExternalRetryInBatchTests; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; -import org.springframework.batch.retry.policy.ItemWriterRetryPolicy; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; import org.springframework.batch.retry.policy.NeverRetryPolicy; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.jms.core.JmsTemplate; import org.springframework.test.AbstractDependencyInjectionSpringContextTests; @@ -44,7 +42,7 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen private int recovered; - private int count; + private volatile int count; /** * Public setter for the {@link BatchMessageListenerContainer}. @@ -101,6 +99,7 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen count++; } }); + container.initializeProxy(); container.start(); jmsTemplate.convertAndSend("queue", "foo"); jmsTemplate.convertAndSend("queue", "bar"); @@ -118,51 +117,40 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen public void onMessage(Message msg) { logger.debug("Message: "+msg); count++; - throw new RuntimeException("planned failure: " + msg); + throw new RuntimeException("planned failure for represent: " + msg); } }); + container.initializeProxy(); container.start(); jmsTemplate.convertAndSend("queue", "foo"); int waiting = 0; - while (count < 2 && waiting++ < 10) { + while (count < 2 && waiting++ < 20) { Thread.sleep(100L); } if (count < 2) { + logger.debug("Count: "+count); fail("Expected message to be processed twice."); } } public void testFailureAndRecovery() throws Exception { final RetryTemplate retryTemplate = new RetryTemplate(); - retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy(new NeverRetryPolicy())); + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(new NeverRetryPolicy())); container.setMessageListener(new MessageListener() { public void onMessage(final Message msg) { try { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback(msg, new AbstractItemWriter() { - public void write(Object item) throws Exception { - logger.debug("Message: "+item); + RecoveryRetryCallback callback = new RecoveryRetryCallback(msg, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + logger.debug("Message: "+msg); count++; throw new RuntimeException("planned failure: " + msg); } - }); - callback.setKeyGenerator(new ItemKeyGenerator() { - public Object getKey(Object item) { - String text; - try { - text = ((TextMessage)item).getJMSMessageID(); - } - catch (JMSException e) { - text = ""+item; - } - logger.debug("Key for message: "+text); - return text; - } - }); - callback.setRecoverer(new ItemRecoverer() { - public boolean recover(Object data, Throwable cause) { + }, msg.getJMSMessageID()); + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable throwable) { recovered++; - logger.debug("Recovered: " + data); - return true; + logger.debug("Recovered: " + msg); + return msg; } }); retryTemplate.execute(callback); @@ -172,6 +160,7 @@ public class BatchMessageListenerContainerIntegrationTests extends AbstractDepen } } }); + container.initializeProxy(); container.start(); jmsTemplate.convertAndSend("queue", "foo"); int waiting = 0; diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java index 488999f41..7c5a998f6 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java @@ -28,10 +28,9 @@ import javax.jms.Session; import junit.framework.TestCase; +import org.aopalliance.aop.Advice; import org.easymock.MockControl; -import org.springframework.batch.container.jms.BatchMessageListenerContainer; -import org.springframework.batch.repeat.RepeatCallback; -import org.springframework.batch.repeat.ExitStatus; +import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.util.ReflectionUtils; @@ -42,27 +41,6 @@ public class BatchMessageListenerContainerTests extends TestCase { int count = 0; - public void testReceiveAndExecuteWithNoCallback() throws Exception { - RepeatTemplate template = new RepeatTemplate() { - public ExitStatus iterate(RepeatCallback callback) { - count++; - return ExitStatus.CONTINUABLE; // means we can continue to operate, but no message is received - } - }; - container = getContainer(template); - boolean received = doExecute(null, null); - assertEquals(1, count); - assertTrue("Message received", received); - } - - private BatchMessageListenerContainer getContainer(RepeatTemplate template) { - MockControl connectionFactoryControl = MockControl.createControl(ConnectionFactory.class); - ConnectionFactory connectionFactory = (ConnectionFactory) connectionFactoryControl.getMock(); - BatchMessageListenerContainer container = new BatchMessageListenerContainer(template); - container.setConnectionFactory(connectionFactory); - return container; - } - public void testReceiveAndExecuteWithCallback() throws Exception { RepeatTemplate template = new RepeatTemplate(); template.setCompletionPolicy(new SimpleCompletionPolicy(2)); @@ -140,7 +118,7 @@ public class BatchMessageListenerContainerTests extends TestCase { container = getContainer(template); container.setSessionTransacted(false); boolean received = doTestWithException(new IllegalStateException("No way!"), false, 1); - assertFalse("Message received successfully", received); + assertTrue("Message not received but listener not transactional so this should be true", received); } public void testNonTransactionalReceiveAndExecuteWithCallbackThrowingError() throws Exception { @@ -150,7 +128,7 @@ public class BatchMessageListenerContainerTests extends TestCase { container.setSessionTransacted(false); try { boolean received = doTestWithException(new RuntimeException("No way!"), false, 1); - assertFalse("Message received successfully", received); + assertTrue("Message not received but listener not transactional so this should be true", received); } catch (RuntimeException e) { assertEquals("No way!", e.getMessage()); @@ -158,6 +136,19 @@ public class BatchMessageListenerContainerTests extends TestCase { } } + private BatchMessageListenerContainer getContainer(RepeatTemplate template) { + MockControl connectionFactoryControl = MockControl.createControl(ConnectionFactory.class); + ConnectionFactory connectionFactory = (ConnectionFactory) connectionFactoryControl.getMock(); + BatchMessageListenerContainer container = new BatchMessageListenerContainer(); + RepeatOperationsInterceptor interceptor = new RepeatOperationsInterceptor(); + interceptor.setRepeatOperations(template); + container.setAdvices(new Advice[] {interceptor}); + container.setConnectionFactory(connectionFactory); + container.setDestinationName("queue"); + container.afterPropertiesSet(); + return container; + } + private boolean doTestWithException(final Throwable t, boolean expectRollback, int expectGetTransactionCount) throws JMSException, IllegalAccessException { container.setAcceptMessagesWhileStopping(true); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java index 14d5ad17d..2345335a0 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java @@ -22,15 +22,17 @@ import java.util.List; import javax.sql.DataSource; import org.springframework.batch.item.AbstractItemReader; -import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; -import org.springframework.batch.retry.policy.ItemWriterRetryPolicy; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; import org.springframework.batch.retry.policy.SimpleRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.jdbc.core.JdbcTemplate; @@ -88,9 +90,9 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring return text; } - public boolean recover(Object data, Throwable cause) { + public Object recover(Object data, Throwable cause) { recovered.add(data); - return true; + return data; } }; retryTemplate = new RetryTemplate(); @@ -113,7 +115,7 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring public void testExternalRetryRecoveryInBatch() throws Exception { assertInitialState(); - retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy(new SimpleRetryPolicy(1))); + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(new SimpleRetryPolicy(1))); repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2)); @@ -129,24 +131,28 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring public ExitStatus doInIteration(RepeatContext context) throws Exception { - Object item = provider.read(); + final Object item = provider.read(); if (item==null) { return ExitStatus.FINISHED; } - ItemWriterRetryCallback callback = new ItemWriterRetryCallback(item, new AbstractItemWriter() { - public void write(final Object text) { + RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { // No need for transaction here: the whole batch will roll // back. When it comes back for recovery this code is not // executed... jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { - new Integer(list.size()), text }); + new Integer(list.size()), item }); throw new RuntimeException("Rollback!"); } }); - callback.setRecoverer(provider); + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable throwable) { + return provider.recover(item, throwable); + } + }); retryTemplate.execute(callback); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java index 3acdaeeb6..edb26de01 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java @@ -73,6 +73,7 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT // Add a couple of messages... jmsTemplate.convertAndSend("queue", "foo"); jmsTemplate.convertAndSend("queue", "bar"); + } protected void onTearDown() throws Exception { @@ -108,6 +109,8 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT } }); + container.initializeProxy(); + container.start(); // Need to sleep for at least a second here... @@ -145,18 +148,20 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT } } }); + + container.initializeProxy(); container.start(); // Need to sleep here, but not too long or the // container goes into its own recovery cycle and spits out the bad // message... - Thread.sleep(500L); + Thread.sleep(1000L); // We rolled back so the messages might come in many times... assertTrue(list.size() >= 1); - System.err.println(jdbcTemplate.queryForList("select * from T_FOOS")); + logger.debug("T_FOOS: "+jdbcTemplate.queryForList("select * from T_FOOS")); String text = ""; List msgs = new ArrayList(); @@ -164,7 +169,7 @@ public class AsynchronousTests extends AbstractDependencyInjectionSpringContextT text = (String) jmsTemplate.receiveAndConvert("queue"); msgs.add(text); } - System.err.println(msgs); + logger.debug("Messages: "+msgs); int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS"); assertEquals(0, count); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java index ff7021daf..be6569f14 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java @@ -100,7 +100,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte System.err.println(jdbcTemplate.queryForList("select * from T_FOOS")); - // Database committed so this resord should be there... + // Database committed so this record should be there... int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS"); assertEquals(2, count); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java index 60cbbd634..78f1c780e 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java @@ -25,8 +25,11 @@ import org.springframework.batch.item.AbstractItemReader; import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.jms.ExternalRetryInBatchTests; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; -import org.springframework.batch.retry.policy.ItemWriterRetryPolicy; +import org.springframework.batch.retry.RecoveryCallback; +import org.springframework.batch.retry.RetryCallback; +import org.springframework.batch.retry.RetryContext; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; +import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jms.core.JmsTemplate; @@ -78,9 +81,9 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext return text; } - public boolean recover(Object data, Throwable cause) { + public Object recover(Object data, Throwable cause) { recovered.add(data); - return true; + return data; } }; retryTemplate = new RetryTemplate(); @@ -105,7 +108,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext assertInitialState(); - retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy()); + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy()); final AbstractItemWriter writer = new AbstractItemWriter() { public void write(final Object text) { @@ -122,7 +125,13 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), writer); + final Object item = provider.read(); + RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + writer.write(item); + return null; + } + }); return retryTemplate.execute(callback); } catch (Exception e) { @@ -144,7 +153,13 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext new TransactionTemplate(transactionManager).execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { try { - ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), writer); + final Object item = provider.read(); + RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { + writer.write(item); + return null; + } + }); return retryTemplate.execute(callback); } catch (Exception e) { @@ -172,16 +187,22 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext assertInitialState(); - retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy()); + retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy()); - final ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() { - public void write(final Object text) { + final Object item = provider.read(); + final RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { - new Integer(list.size()), text }); + new Integer(list.size()), item }); throw new RuntimeException("Rollback!"); } }); - callback.setRecoverer(provider); + + callback.setRecoveryCallback(new RecoveryCallback() { + public Object recover(Throwable throwable) { + return provider.recover(item, throwable); + } + }); Object result = "start"; diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java index ba9db883a..0b61430d9 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java @@ -19,12 +19,11 @@ package org.springframework.batch.retry.jms; import java.util.ArrayList; import java.util.List; -import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.jms.JmsItemReader; import org.springframework.batch.jms.ExternalRetryInBatchTests; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.jms.core.JmsTemplate; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -150,23 +149,24 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte provider.setJmsTemplate(jmsTemplate); jmsTemplate.setDefaultDestinationName("queue"); - retryTemplate.execute(new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() { - public void write(final Object text) { + final Object item = provider.read(); + retryTemplate.execute(new RecoveryRetryCallback(item, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED); - transactionTemplate.execute(new TransactionCallback() { + return transactionTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - list.add(text); - System.err.println("Inserting: [" + list.size() + "," + text + "]"); + list.add(item); + System.err.println("Inserting: [" + list.size() + "," + item + "]"); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { - new Integer(list.size()), text }); + new Integer(list.size()), item }); if (list.size() == 1) { throw new RuntimeException("Rollback!"); } - return text; + return item; } }); diff --git a/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java index e68363f43..96e13ef8a 100644 --- a/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java @@ -19,12 +19,11 @@ package org.springframework.retry.jms; import java.util.ArrayList; import java.util.List; -import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.jms.JmsItemReader; import org.springframework.batch.jms.ExternalRetryInBatchTests; import org.springframework.batch.retry.RetryCallback; import org.springframework.batch.retry.RetryContext; -import org.springframework.batch.retry.callback.ItemWriterRetryCallback; +import org.springframework.batch.retry.callback.RecoveryRetryCallback; import org.springframework.batch.retry.support.RetryTemplate; import org.springframework.jms.core.JmsTemplate; import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests; @@ -53,6 +52,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte String foo = ""; int count = 0; while (foo != null && count < 100) { + logger.debug("Drained message: "+count+": "+foo); foo = (String) jmsTemplate.receiveAndConvert("queue"); count++; } @@ -61,10 +61,6 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte retryTemplate = new RetryTemplate(); } - protected void onSetUpInTransaction() throws Exception { - super.onSetUpInTransaction(); - } - private void assertInitialState() { int count = jdbcTemplate.queryForInt("select count(*) from T_FOOS"); assertEquals(0, count); @@ -101,7 +97,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte public Object doInTransaction(TransactionStatus status) { list.add(text); - System.err.println("Inserting: [" + list.size() + "," + text + "]"); + logger.debug("Inserting: [" + list.size() + "," + text + "]"); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(list.size()), text }); if (list.size() == 1) { @@ -146,16 +142,17 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte provider.setJmsTemplate(jmsTemplate); jmsTemplate.setDefaultDestinationName("queue"); - retryTemplate.execute(new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() { - public void write(final Object text) { + final Object text = provider.read(); + retryTemplate.execute(new RecoveryRetryCallback(text, new RetryCallback() { + public Object doWithRetry(RetryContext context) throws Throwable { TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); transactionTemplate.setPropagationBehavior(TransactionTemplate.PROPAGATION_NESTED); - transactionTemplate.execute(new TransactionCallback() { + return transactionTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { list.add(text); - System.err.println("Inserting: [" + list.size() + "," + text + "]"); + logger.debug("Inserting: [" + list.size() + "," + text + "]"); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(list.size()), text }); if (list.size() == 1) { @@ -215,7 +212,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte public Object doInTransaction(TransactionStatus status) { list.add(text); - System.err.println("Inserting: [" + list.size() + "," + text + "]"); + logger.debug("Inserting: [" + list.size() + "," + text + "]"); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(list.size()), text }); return text; @@ -271,6 +268,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte // transaction... final String text = (String) jmsTemplate.receiveAndConvert("queue"); list.add(text); + logger.debug("Processing Foo: "+text); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(list.size()), text }); if (list.size() == 1) { @@ -319,10 +317,11 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte return transactionTemplate.execute(new TransactionCallback() { public Object doInTransaction(TransactionStatus status) { - // The receieve is inside the retry and the + // The receive is inside the retry and the // transaction... final String text = (String) jmsTemplate.receiveAndConvert("queue"); list.add(text); + logger.debug("Processing Foo: "+text); jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] { new Integer(list.size()), text }); throw new RuntimeException("Rollback!"); diff --git a/spring-batch-integration/src/test/resources/log4j.properties b/spring-batch-integration/src/test/resources/log4j.properties index b31319400..f8c8855eb 100644 --- a/spring-batch-integration/src/test/resources/log4j.properties +++ b/spring-batch-integration/src/test/resources/log4j.properties @@ -6,4 +6,7 @@ log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{1}:%L - %m log4j.category.org.apache.activemq=ERROR # log4j.category.org.springframework=DEBUG -log4j.category.org.springframework.batch.container.jms=DEBUG +log4j.category.org.springframework.jdbc=DEBUG +log4j.category.org.springframework.jms=DEBUG +log4j.category.org.springframework.batch=DEBUG +log4j.category.org.springframework.retry=DEBUG diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/jms/jms-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/jms/jms-context.xml index 921cb6d37..ee2d6d232 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/jms/jms-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/jms/jms-context.xml @@ -1,19 +1,16 @@ - - + - + @@ -23,21 +20,18 @@ - + - + - + vm://localhost @@ -45,28 +39,43 @@ - + - + + + + + - - + - + + + + + + + + + + + + - + - + @@ -76,20 +85,14 @@ class="org.springframework.transaction.interceptor.TransactionProxyFactoryBean"> - - org.springframework.batch.repeat.RepeatOperations - + org.springframework.batch.repeat.RepeatOperations - - + - + - + diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java index 240b32442..c8c4e92d8 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java @@ -52,8 +52,8 @@ public class GeneratingItemReader implements ItemReader, ItemRecoverer { /* (non-Javadoc) * @see org.springframework.batch.item.ItemRecoverer#recover(java.lang.Object, java.lang.Throwable) */ - public boolean recover(Object data, Throwable cause) { - return false; + public Object recover(Object data, Throwable cause) { + return data; } /* (non-Javadoc)