Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

OPEN - issue BATCH-569: Add RetryOperationsInterceptor with stateful retry

Browse Source 
Refactored ItemWriterRetry* to be less dependent on Item* interfaces.  Added StatefulRetryOperationsInterceptor.
This commit is contained in:
dsyer
2008-05-09 20:59:59 +00:00
parent b09b30b158
commit a0b7049867
32 changed files with 1165 additions and 731 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java
View File

@@ -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);
}

4
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
View File

@@ -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();

2
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java
View File

@@ -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);
}

13
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/FailedItemIdentifier.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/item/NewItemIdentifier.java
View File

@@ -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);
}

34
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/jms/JmsItemReader.java
View File

@@ -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;
}
}

164
spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/interceptor/RepeatOperationsInterceptor.java
View File

@@ -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);
}
}

31
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/RecoveryCallback.java Normal file
View File

@@ -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);
}

174
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java
View File

@@ -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;
}
}

139
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/RecoveryRetryCallback.java Normal file
View File

@@ -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;
}
}

141
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptor.java Normal file
View File

@@ -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;
}
}

76
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java → spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/RecoveryCallbackRetryPolicy.java
View File

@@ -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.<br/>
*
* 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);
}
}

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java
View File

@@ -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;
}
};

4
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/jms/JmsItemReaderTests.java
View File

@@ -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"));
}
}

14
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/StubItemKeyGeneratorRecoverer.java
View File

@@ -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)

48
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java → spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/RecoveryRetryCallbackTests.java
View File

@@ -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 {

212
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/interceptor/StatefulRetryOperationsInterceptorTests.java Normal file
View File

@@ -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);
}
}
}

210
spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java → spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/RecoveryRetryPolicyTests.java
View File

@@ -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;

9
spring-batch-integration/.springBeans
View File

@@ -1,8 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?>
<beansProjectDescription>
<configExtensions>
<configExtension>xml</configExtension>
</configExtensions>
<version>1</version>
<pluginVersion><![CDATA[2.0.5.v200805010412]]></pluginVersion>
<configSuffixes>
<configSuffix><![CDATA[xml]]></configSuffix>
</configSuffixes>
<enableImports><![CDATA[false]]></enableImports>
<configs>
<config>src/test/resources/org/springframework/batch/jms/jms-context.xml</config>
<config>src/test/resources/data-source.xml</config>

1
spring-batch-integration/pom.xml
View File

@@ -200,7 +200,6 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

138
spring-batch-integration/spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java Normal file
View File

@@ -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.<br/>
*
* 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);
}
}

137
spring-batch-integration/src/main/java/org/springframework/batch/container/jms/BatchMessageListenerContainer.java
View File

@@ -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;
}
}

53
spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java
View File

@@ -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;

43
spring-batch-integration/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerTests.java
View File

@@ -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);

28
spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java
View File

@@ -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);

11
spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java
View File

@@ -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);

2
spring-batch-integration/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java
View File

@@ -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);

45
spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java
View File

@@ -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";

18
spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java
View File

@@ -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;
}
});

25
spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java
View File

@@ -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!");

5
spring-batch-integration/src/test/resources/log4j.properties
View File

@@ -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

71
spring-batch-integration/src/test/resources/org/springframework/batch/jms/jms-context.xml
View File

@@ -1,19 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:tx="http://www.springframework.org/schema/tx"
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:p="http://www.springframework.org/schema/p" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd">
<import resource="classpath:/data-source.xml"/>
<import resource="classpath:/data-source.xml" />
<!-- Transaction manager for a datasource -->
<bean id="transactionManager"
class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
</bean>
@@ -23,21 +20,18 @@
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<bean id="jmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="receiveTimeout" value="100" />
<!-- This is important... -->
<property name="sessionTransacted" value="true" />
</bean>
<bean id="jdbcTemplate"
class="org.springframework.jdbc.core.JdbcTemplate">
<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="connectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="brokerService">
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="brokerService">
<property name="brokerURL">
<value>vm://localhost</value>
</property>
@@ -45,28 +39,43 @@
<bean id="txAwareConnectionFactory"
class="org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy">
<property name="targetConnectionFactory" ref="connectionFactory"/>
<property name="targetConnectionFactory" ref="connectionFactory" />
<property name="synchedLocalTransactionAllowed" value="true" />
</bean>
<bean id="container"
class="org.springframework.batch.container.jms.BatchMessageListenerContainer">
<bean id="container" class="org.springframework.batch.container.jms.BatchMessageListenerContainer"
lazy-init="true">
<property name="recoveryInterval" value="500" />
<!-- We aren't adding a listener here (want to do it in unit test)
so need to set autoStartup=false -->
<property name="autoStartup" value="false" />
<!-- We need the transaction manager, but only because we can't
intercept the MessageListenerContainer in the right place
(it tries to open a JMS session before receiveAndExecute()
so the JMS session commits after every item, not after every chunk) -->
<property name="transactionManager" ref="transactionManager" />
<property name="recoveryInterval" value="0" />
<property name="connectionFactory"
ref="txAwareConnectionFactory" />
<property name="connectionFactory" ref="txAwareConnectionFactory" />
<property name="destinationName" value="queue" />
<!-- This is important... it forces the container to acknowledge message receipt,
and avoid duplicate messages in the sunny day case -->
<property name="sessionTransacted" value="true" />
<constructor-arg ref="transactionalBatchTemplate" />
<property name="advices">
<list>
<!-- Need another transaction manager here so that rollbacks are processed correctly -->
<bean class="org.springframework.transaction.interceptor.TransactionInterceptor">
<property name="transactionManager" ref="transactionManager" />
<property name="transactionAttributes" value="*=PROPAGATION_REQUIRED" />
</bean>
<bean class="org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor">
<property name="repeatOperations" ref="batchTemplate" />
</bean>
</list>
</property>
</bean>
<bean id="batchTemplate"
class="org.springframework.batch.repeat.support.RepeatTemplate">
<bean id="batchTemplate" class="org.springframework.batch.repeat.support.RepeatTemplate">
<property name="completionPolicy">
<bean
class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<bean class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<constructor-arg value="2" />
</bean>
</property>
@@ -76,20 +85,14 @@
class="org.springframework.transaction.interceptor.TransactionProxyFactoryBean">
<property name="transactionManager" ref="transactionManager" />
<property name="proxyInterfaces">
<value>
org.springframework.batch.repeat.RepeatOperations
</value>
<value>org.springframework.batch.repeat.RepeatOperations</value>
</property>
<property name="proxyTargetClass" value="false" />
<property name="transactionAttributes"
value="*=PROPAGATION_REQUIRED">
</property>
<property name="transactionAttributes" value="*=PROPAGATION_REQUIRED"></property>
<property name="target">
<bean
class="org.springframework.batch.repeat.support.RepeatTemplate">
<bean class="org.springframework.batch.repeat.support.RepeatTemplate">
<property name="completionPolicy">
<bean
class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<bean class="org.springframework.batch.repeat.policy.SimpleCompletionPolicy">
<constructor-arg value="2" />
</bean>
</property>

4
spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java
View File

@@ -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)