OPEN - issue BATCH-777: Parametrise RetryCallback and related interfaces
Make stateful retry explicit in the RepeatOperations interface instead of hidden in a RetryPolicy
This commit is contained in:
dsyer
@@ -38,7 +38,8 @@ public interface RetryOperations {
|
||||
|
||||
/**
|
||||
* Execute the supplied {@link RetryCallback} with a fallback on exhausted
|
||||
* retry to the {@link RecoveryCallback}. See implementations for configuration details.
|
||||
* retry to the {@link RecoveryCallback}. See implementations for
|
||||
* configuration details.
|
||||
*
|
||||
* @return the value returned by the {@link RetryCallback} upon successful
|
||||
* invocation, and that returned by the {@link RecoveryCallback} otherwise.
|
||||
@@ -47,4 +48,40 @@ public interface RetryOperations {
|
||||
*/
|
||||
Object execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback) throws Exception;
|
||||
|
||||
/**
|
||||
* A simple stateful retry. Execute the supplied {@link RetryCallback} with
|
||||
* a target object for the attempt identified by the {@link RetryState}.
|
||||
* Exceptions thrown by the callback are always propagated immediately so
|
||||
* the state is required to be able to identify the previous attempt, if
|
||||
* there is one - hence the state is required. Normal patterns would see
|
||||
* this method being used inside a transaction, where the callback might
|
||||
* invalidate the transaction if it fails.<br/><br/>
|
||||
*
|
||||
* See implementations for configuration details.
|
||||
*
|
||||
* @return the value returned by the {@link RetryCallback} upon successful
|
||||
* invocation, and that returned by the {@link RecoveryCallback} otherwise.
|
||||
* @throws Exception any {@link Exception} raised by the
|
||||
* {@link RecoveryCallback}.
|
||||
* @throws ExhaustedRetryException if the last attempt for this state has
|
||||
* already been reached
|
||||
*/
|
||||
Object execute(RetryCallback retryCallback, RetryState retryState) throws Exception, ExhaustedRetryException;
|
||||
|
||||
/**
|
||||
* A stateful retry with a recovery path. Execute the supplied
|
||||
* {@link RetryCallback} with a fallback on exhausted retry to the
|
||||
* {@link RecoveryCallback} and a target object for the retry attempt
|
||||
* identified by the {@link RetryState}.
|
||||
*
|
||||
* @see #execute(RetryCallback, RetryState)
|
||||
*
|
||||
* @return the value returned by the {@link RetryCallback} upon successful
|
||||
* invocation, and that returned by the {@link RecoveryCallback} otherwise.
|
||||
* @throws Exception any {@link Exception} raised by the
|
||||
* {@link RecoveryCallback} upon unsuccessful retry.
|
||||
*/
|
||||
Object execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, RetryState retryState)
|
||||
throws Exception;
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
/*
|
||||
* 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 class RetryState {
|
||||
|
||||
final private Object key;
|
||||
final private boolean forceRefresh;
|
||||
|
||||
/**
|
||||
* @param key
|
||||
* @param forceRefresh
|
||||
*/
|
||||
public RetryState(Object key, boolean forceRefresh) {
|
||||
this.key = key;
|
||||
this.forceRefresh = forceRefresh;
|
||||
}
|
||||
|
||||
public RetryState(Object key) {
|
||||
this(key, false);
|
||||
}
|
||||
/**
|
||||
* @return the key that this state represents
|
||||
*/
|
||||
public Object getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return true if the state requires an explicit check for the key
|
||||
*/
|
||||
public boolean isForceRefresh() {
|
||||
return forceRefresh;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,132 +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.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.policy.RecoveryCallbackRetryPolicy;
|
||||
|
||||
/**
|
||||
* A {@link RetryCallback} that knows about and caches an item, and attempts to
|
||||
* process it using a delegate {@link RetryCallback}. Used by the
|
||||
* {@link RecoveryCallbackRetryPolicy} to enable stateful retry of the
|
||||
* 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 callback the delegate to use to process it
|
||||
*/
|
||||
public RecoveryRetryCallback(Object item, RetryCallback callback) {
|
||||
super();
|
||||
this.item = item;
|
||||
this.callback = callback;
|
||||
this.key = item;
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructor with mandatory properties.
|
||||
*
|
||||
* @param item the item to process
|
||||
* @param callback the delegate to use to process it
|
||||
*/
|
||||
public RecoveryRetryCallback(Object item, RetryCallback callback, Object key) {
|
||||
super();
|
||||
this.item = item;
|
||||
this.callback = callback;
|
||||
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.
|
||||
*
|
||||
* @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 Exception {
|
||||
return callback.doWithRetry(context);
|
||||
// N.B. code used to check here for isExhaustedOnly and throw exception.
|
||||
// This is unnecessary because the callback could just throw the
|
||||
// exception itself if it wants to go to the recovery path.
|
||||
}
|
||||
|
||||
public Object getItem() {
|
||||
return item;
|
||||
}
|
||||
|
||||
public boolean isForceRefresh() {
|
||||
return forceRefresh;
|
||||
}
|
||||
|
||||
/**
|
||||
* Accessor for the {@link RecoveryCallback}.
|
||||
*
|
||||
* @return the {@link RecoveryCallback}.
|
||||
*/
|
||||
public RecoveryCallback getRecoveryCallback() {
|
||||
return recoverer;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -29,9 +29,8 @@ import org.springframework.batch.retry.RetryCallback;
|
||||
import org.springframework.batch.retry.RetryContext;
|
||||
import org.springframework.batch.retry.RetryOperations;
|
||||
import org.springframework.batch.retry.RetryPolicy;
|
||||
import org.springframework.batch.retry.callback.RecoveryRetryCallback;
|
||||
import org.springframework.batch.retry.RetryState;
|
||||
import org.springframework.batch.retry.policy.NeverRetryPolicy;
|
||||
import org.springframework.batch.retry.policy.RecoveryCallbackRetryPolicy;
|
||||
import org.springframework.batch.retry.support.RetryTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
@@ -71,7 +70,7 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
|
||||
*/
|
||||
public StatefulRetryOperationsInterceptor() {
|
||||
super();
|
||||
retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(new NeverRetryPolicy()));
|
||||
retryTemplate.setRetryPolicy(new NeverRetryPolicy());
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -100,7 +99,7 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
|
||||
* @param retryPolicy the retryPolicy to set
|
||||
*/
|
||||
public void setRetryPolicy(RetryPolicy retryPolicy) {
|
||||
retryTemplate.setRetryPolicy(new RecoveryCallbackRetryPolicy(retryPolicy));
|
||||
retryTemplate.setRetryPolicy(retryPolicy);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -142,14 +141,9 @@ public class StatefulRetryOperationsInterceptor implements MethodInterceptor {
|
||||
}
|
||||
final Object item = arg;
|
||||
|
||||
RecoveryRetryCallback callback = new RecoveryRetryCallback(item, new MethodInvocationRetryCallback(invocation),
|
||||
keyGenerator != null ? keyGenerator.getKey(item) : item);
|
||||
callback.setRecoveryCallback(new ItemRecovererCallback(item, recoverer));
|
||||
if (newItemIdentifier != null) {
|
||||
callback.setForceRefresh(newItemIdentifier.isNew(item));
|
||||
}
|
||||
RetryState retryState = new RetryState(keyGenerator != null ? keyGenerator.getKey(item) : item, newItemIdentifier != null ? newItemIdentifier.isNew(item) : false );
|
||||
|
||||
Object result = retryTemplate.execute(callback);
|
||||
Object result = retryTemplate.execute(new MethodInvocationRetryCallback(invocation), new ItemRecovererCallback(item, recoverer), retryState);
|
||||
|
||||
logger.debug("Exiting proxied method in stateful retry with result: (" + result + ")");
|
||||
|
||||
|
||||
@@ -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.retry.policy;
|
||||
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
|
||||
import org.springframework.batch.retry.ExhaustedRetryException;
|
||||
import org.springframework.batch.retry.RetryContext;
|
||||
import org.springframework.batch.retry.RetryPolicy;
|
||||
|
||||
/**
|
||||
* Base class for stateful retry policies: those that operate in the context of
|
||||
* a callback that is called once per retry execution (usually to enforce that
|
||||
* it is only called once per transaction). Stateful policies need to remember
|
||||
* the context for the operation that failed (e.g. the data item that was being
|
||||
* processed), and decide based on its history what to do in the current
|
||||
* context. For example: the retry operation includes receiving a message, and
|
||||
* we need it to roll back and be re-delivered so that we can have another crack
|
||||
* at it.
|
||||
*
|
||||
* @see RetryPolicy#handleRetryExhausted(RetryContext)
|
||||
* @see AbstractStatelessRetryPolicy
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractStatefulRetryPolicy implements RetryPolicy {
|
||||
|
||||
private volatile Set<Class<?>> recoverableExceptionClasses = new HashSet<Class<?>>();
|
||||
|
||||
protected RetryContextCache retryContextCache = new MapRetryContextCache();
|
||||
|
||||
/**
|
||||
* Optional setter for the retry context cache. The default value is a
|
||||
* {@link MapRetryContextCache}.
|
||||
*
|
||||
* @param retryContextCache
|
||||
*/
|
||||
public void setRetryContextCache(RetryContextCache retryContextCache) {
|
||||
this.retryContextCache = retryContextCache;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return null. Subclasses should provide a recovery path if possible.
|
||||
* Subclasses are also encouraged not to declare throws Exception if they
|
||||
* can (e.g. in the plausible and common case that the recovery is a last
|
||||
* ditch effort to prevent a message going back to the middleware, for
|
||||
* instance). Any subclass that actually does throw an Exception of any type
|
||||
* should be aware that it will simply be propagated and the caller will
|
||||
* have top deal with it.
|
||||
*
|
||||
* @throws Exception if the recovery path demands it
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryPolicy#handleRetryExhausted(org.springframework.batch.retry.RetryContext)
|
||||
*/
|
||||
public Object handleRetryExhausted(RetryContext context) throws ExhaustedRetryException, Exception {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* For a stateful policy the default is to always rethrow. This is the
|
||||
* cautious approach: we assume that the failed processing may have written
|
||||
* data to a transactional resource, so we rethrow and force a rollback. Any
|
||||
* recovery path that may be available has to be taken on the next attempt,
|
||||
* before any processing has taken place.
|
||||
*
|
||||
* @return true unless the last exception registered was recoverable.
|
||||
*/
|
||||
public boolean shouldRethrow(RetryContext context) {
|
||||
return !recoverForException(context.getLastThrowable());
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the recoverable exceptions. Any exception on the list, or subclasses
|
||||
* thereof, will be recoverable. If it is encountered in a retry block it
|
||||
* will not be rethrown. Others will be rethrown. The recovery action (if
|
||||
* any) is left to subclasses - normally they would override
|
||||
* {@link #handleRetryExhausted(RetryContext)}.
|
||||
*
|
||||
* @param retryableExceptionClasses defaults to {@link Exception}.
|
||||
*/
|
||||
public final void setRecoverableExceptionClasses(Class<?>[] retryableExceptionClasses) {
|
||||
Set<Class<?>> temp = new HashSet<Class<?>>();
|
||||
for (int i = 0; i < retryableExceptionClasses.length; i++) {
|
||||
addRecoverableExceptionClass(retryableExceptionClasses[i], temp);
|
||||
}
|
||||
this.recoverableExceptionClasses = temp;
|
||||
}
|
||||
|
||||
private void addRecoverableExceptionClass(Class<?> retryableExceptionClass, Set<Class<?>> set) {
|
||||
if (!Throwable.class.isAssignableFrom(retryableExceptionClass)) {
|
||||
throw new IllegalArgumentException("Class '" + retryableExceptionClass.getName()
|
||||
+ "' is not a subtype of Throwable.");
|
||||
}
|
||||
set.add(retryableExceptionClass);
|
||||
}
|
||||
|
||||
protected boolean recoverForException(Throwable ex) {
|
||||
|
||||
// Default is false (but this shouldn't really happen in practice -
|
||||
// maybe in tests):
|
||||
if (ex == null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
Class<? extends Throwable> exceptionClass = ex.getClass();
|
||||
if (recoverableExceptionClasses.contains(exceptionClass)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// check for subclasses
|
||||
for (Class<?> cls : recoverableExceptionClasses) {
|
||||
if (cls.isAssignableFrom(exceptionClass)) {
|
||||
addRecoverableExceptionClass(exceptionClass, this.recoverableExceptionClasses);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -26,10 +26,11 @@ import org.springframework.batch.retry.RetryPolicy;
|
||||
* state outside the context.
|
||||
*
|
||||
* @see RetryPolicy#handleRetryExhausted(RetryContext)
|
||||
* @see AbstractStatefulRetryPolicy
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
* @deprecated TODO: remove this base class
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractStatelessRetryPolicy implements RetryPolicy {
|
||||
|
||||
|
||||
@@ -1,226 +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.policy;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.retry.ExhaustedRetryException;
|
||||
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.RecoveryRetryCallback;
|
||||
import org.springframework.batch.retry.context.RetryContextSupport;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* 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 RecoveryRetryCallback} otherwise an exception will be thrown when
|
||||
* the context is created.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class RecoveryCallbackRetryPolicy extends AbstractStatefulRetryPolicy {
|
||||
|
||||
protected Log logger = LogFactory.getLog(getClass());
|
||||
|
||||
public static final String EXHAUSTED = RecoveryCallbackRetryPolicy.class.getName() + ".EXHAUSTED";
|
||||
|
||||
private RetryPolicy delegate;
|
||||
|
||||
/**
|
||||
* Convenience constructor to set delegate on init.
|
||||
*
|
||||
* @param delegate
|
||||
*/
|
||||
public RecoveryCallbackRetryPolicy(RetryPolicy delegate) {
|
||||
super();
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Default constructor. Creates a new {@link SimpleRetryPolicy} for the
|
||||
* delegate.
|
||||
*/
|
||||
public RecoveryCallbackRetryPolicy() {
|
||||
this(new SimpleRetryPolicy());
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for delegate.
|
||||
*
|
||||
* @param delegate
|
||||
*/
|
||||
public void setDelegate(RetryPolicy delegate) {
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check the history of this item, and if it has reached the retry limit,
|
||||
* then return false.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryPolicy#canRetry(org.springframework.batch.retry.RetryContext)
|
||||
*/
|
||||
public boolean canRetry(RetryContext context) {
|
||||
return ((RetryPolicy) context).canRetry(context);
|
||||
}
|
||||
|
||||
/**
|
||||
* Delegates to the delegate context.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryPolicy#close(org.springframework.batch.retry.RetryContext, boolean)
|
||||
*/
|
||||
public void close(RetryContext context, boolean succeeded) {
|
||||
((RetryPolicy) context).close(context, succeeded);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new context for the execution of the callback, which must be an
|
||||
* instance of {@link RecoveryRetryCallback}.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback,
|
||||
* RetryContext)
|
||||
*
|
||||
* @throws IllegalStateException if the callback is not of the required
|
||||
* type.
|
||||
*/
|
||||
public RetryContext open(RetryCallback callback, RetryContext parent) {
|
||||
Assert.state(callback instanceof RecoveryRetryCallback, "Callback must be RecoveryRetryCallback");
|
||||
RecoveryCallbackRetryContext context = new RecoveryCallbackRetryContext((RecoveryRetryCallback) callback, parent);
|
||||
context.open(callback, null);
|
||||
return context;
|
||||
}
|
||||
|
||||
/**
|
||||
* If {@link #canRetry(RetryContext)} is false then take remedial action (if
|
||||
* implemented by subclasses), and remove the current item from the history.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryPolicy#registerThrowable(org.springframework.batch.retry.RetryContext,
|
||||
* Exception)
|
||||
*/
|
||||
public void registerThrowable(RetryContext context, Exception throwable) throws TerminatedRetryException {
|
||||
((RetryPolicy) context).registerThrowable(context, throwable);
|
||||
// The throwable is stored in the delegate context.
|
||||
}
|
||||
|
||||
/**
|
||||
* Call recovery path (if any) and clean up context history.
|
||||
*
|
||||
* @see org.springframework.batch.retry.policy.AbstractStatefulRetryPolicy#handleRetryExhausted(org.springframework.batch.retry.RetryContext)
|
||||
*/
|
||||
public Object handleRetryExhausted(RetryContext context) throws Exception, ExhaustedRetryException {
|
||||
return ((RetryPolicy) context).handleRetryExhausted(context);
|
||||
}
|
||||
|
||||
private class RecoveryCallbackRetryContext extends RetryContextSupport implements RetryPolicy {
|
||||
|
||||
final private Object key;
|
||||
|
||||
final private int initialHashCode;
|
||||
|
||||
// The delegate context...
|
||||
private RetryContext delegateContext;
|
||||
|
||||
final private RecoveryCallback recoverer;
|
||||
|
||||
final private boolean forceRefresh;
|
||||
|
||||
public RecoveryCallbackRetryContext(RecoveryRetryCallback callback, RetryContext parent) {
|
||||
super(parent);
|
||||
this.recoverer = callback.getRecoveryCallback();
|
||||
this.key = callback.getKey();
|
||||
this.forceRefresh = callback.isForceRefresh();
|
||||
this.initialHashCode = key.hashCode();
|
||||
}
|
||||
|
||||
public boolean canRetry(RetryContext context) {
|
||||
return delegate.canRetry(this.delegateContext);
|
||||
}
|
||||
|
||||
public void close(RetryContext context, boolean succeeded) {
|
||||
if (succeeded) {
|
||||
retryContextCache.remove(key);
|
||||
delegate.close(this.delegateContext, succeeded);
|
||||
}
|
||||
}
|
||||
|
||||
public RetryContext open(RetryCallback callback, RetryContext parent) {
|
||||
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. "
|
||||
+ "Consider whether equals() or hashCode() for the item might be inconsistent, "
|
||||
+ "or if you need to supply a better ItemKeyGenerator");
|
||||
}
|
||||
}
|
||||
else {
|
||||
// Only create a new context if we don't know the history of
|
||||
// this item:
|
||||
this.delegateContext = delegate.open(callback, null);
|
||||
}
|
||||
// The return value shouldn't be used...
|
||||
return null;
|
||||
}
|
||||
|
||||
public void registerThrowable(RetryContext context, Exception throwable) throws TerminatedRetryException {
|
||||
// TODO: this comparison assumes that hashCode is the limiting
|
||||
// factor. Actually the cache should be able to decide for us.
|
||||
if (this.initialHashCode != key.hashCode()) {
|
||||
throw new RetryException("Inconsistent state for failed item key: hashCode has changed. "
|
||||
+ "Consider whether equals() or hashCode() for the item might be inconsistent, "
|
||||
+ "or if you need to supply a better ItemKeyGenerator");
|
||||
}
|
||||
retryContextCache.put(key, this.delegateContext);
|
||||
delegate.registerThrowable(this.delegateContext, throwable);
|
||||
}
|
||||
|
||||
public boolean shouldRethrow(RetryContext context) {
|
||||
// Not called...
|
||||
throw new UnsupportedOperationException("Not supported - this code should be unreachable.");
|
||||
}
|
||||
|
||||
public Object handleRetryExhausted(RetryContext context) throws Exception, ExhaustedRetryException {
|
||||
// If there is no going back, then we can remove the history
|
||||
retryContextCache.remove(key);
|
||||
if (recoverer != null) {
|
||||
return recoverer.recover(context);
|
||||
}
|
||||
logger.info("No recovery callback provided. Returning null from recovery step.");
|
||||
// Don't want to call the delegate here - it would throw an exception
|
||||
return null;
|
||||
}
|
||||
|
||||
public Exception getLastThrowable() {
|
||||
return delegateContext.getLastThrowable();
|
||||
}
|
||||
|
||||
public int getRetryCount() {
|
||||
return delegateContext.getRetryCount();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -98,9 +98,6 @@ public class SimpleRetryPolicy extends AbstractStatelessRetryPolicy {
|
||||
public boolean canRetry(RetryContext context) {
|
||||
SimpleRetryContext simpleContext = ((SimpleRetryContext) context);
|
||||
Throwable t = simpleContext.getLastThrowable();
|
||||
// N.B. since the contract is defined to include the initial attempt
|
||||
// in the count, we have to subtract one from the max attempts in this
|
||||
// test
|
||||
return (t == null || retryForException(t)) && simpleContext.getRetryCount() < maxAttempts;
|
||||
}
|
||||
|
||||
|
||||
@@ -26,14 +26,18 @@ import org.springframework.batch.retry.ExhaustedRetryException;
|
||||
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.RetryState;
|
||||
import org.springframework.batch.retry.TerminatedRetryException;
|
||||
import org.springframework.batch.retry.backoff.BackOffContext;
|
||||
import org.springframework.batch.retry.backoff.BackOffInterruptedException;
|
||||
import org.springframework.batch.retry.backoff.BackOffPolicy;
|
||||
import org.springframework.batch.retry.backoff.NoBackOffPolicy;
|
||||
import org.springframework.batch.retry.policy.MapRetryContextCache;
|
||||
import org.springframework.batch.retry.policy.RetryContextCache;
|
||||
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
|
||||
|
||||
/**
|
||||
@@ -72,6 +76,16 @@ public class RetryTemplate implements RetryOperations {
|
||||
|
||||
private volatile RetryListener[] listeners = new RetryListener[0];
|
||||
|
||||
private RetryContextCache retryContextCache = new MapRetryContextCache();
|
||||
|
||||
/**
|
||||
* Public setter for the {@link RetryContextCache}.
|
||||
* @param retryContextCache the {@link RetryContextCache} to set.
|
||||
*/
|
||||
public void setRetryContextCache(RetryContextCache retryContextCache) {
|
||||
this.retryContextCache = retryContextCache;
|
||||
}
|
||||
|
||||
/**
|
||||
* Setter for listeners. The listeners are executed before and after a retry
|
||||
* block (i.e. before and after all the attempts), and on an error (every
|
||||
@@ -121,9 +135,8 @@ public class RetryTemplate implements RetryOperations {
|
||||
* @throws TerminatedRetryException if the retry has been manually
|
||||
* terminated through the {@link RetryContext}.
|
||||
*/
|
||||
public final Object execute(RetryCallback callback) throws Exception {
|
||||
RetryPolicy retryPolicy = this.retryPolicy;
|
||||
return doExecute(callback, null, retryPolicy);
|
||||
public final Object execute(RetryCallback retryCallback) throws Exception {
|
||||
return doExecute(retryCallback, null, null);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -138,24 +151,52 @@ public class RetryTemplate implements RetryOperations {
|
||||
* terminated through the {@link RetryContext}.
|
||||
*/
|
||||
public final Object execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback) throws Exception {
|
||||
RetryPolicy retryPolicy = this.retryPolicy;
|
||||
return doExecute(retryCallback, recoveryCallback, retryPolicy);
|
||||
return doExecute(retryCallback, recoveryCallback, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param retryCallback
|
||||
* @param recoveryCallback
|
||||
* @param retryPolicy
|
||||
* @return the result of the callback
|
||||
* @throws Exception
|
||||
* Execute the callback once if the policy dictates that we can, re-throwing
|
||||
* any exception encountered.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryOperations#execute(RetryCallback,
|
||||
* RetryState)
|
||||
*
|
||||
* @throws ExhaustedRetryException if the retry has been exhausted.
|
||||
*/
|
||||
protected Object doExecute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, RetryPolicy retryPolicy)
|
||||
throws Exception {
|
||||
public final Object execute(RetryCallback retryCallback, RetryState retryState) throws Exception,
|
||||
ExhaustedRetryException {
|
||||
return doExecute(retryCallback, null, retryState);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the callback once if the policy dictates that we can, re-throwing
|
||||
* any exception encountered.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryOperations#execute(RetryCallback,
|
||||
* RetryState)
|
||||
*/
|
||||
public final Object execute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, RetryState retryState)
|
||||
throws Exception, ExhaustedRetryException {
|
||||
return doExecute(retryCallback, recoveryCallback, retryState);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the callback once if the policy dictates that we can, otherwise
|
||||
* execute the recovery callback.
|
||||
*
|
||||
* @see org.springframework.batch.retry.RetryOperations#execute(RetryCallback,
|
||||
* RecoveryCallback, RetryState)
|
||||
* @throws ExhaustedRetryException if the retry has been exhausted.
|
||||
*/
|
||||
protected Object doExecute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, RetryState state)
|
||||
throws Exception, ExhaustedRetryException {
|
||||
|
||||
RetryPolicy retryPolicy = this.retryPolicy;
|
||||
BackOffPolicy backOffPolicy = this.backOffPolicy;
|
||||
|
||||
// Allow the retry policy to initialise itself...
|
||||
// TODO: catch and rethrow abnormal retry exception?
|
||||
RetryContext context = retryPolicy.open(retryCallback, RetrySynchronizationManager.getContext());
|
||||
RetryContext context = open(retryCallback, retryPolicy, state);
|
||||
|
||||
// Make sure the context is available globally for clients who need
|
||||
// it...
|
||||
@@ -196,9 +237,9 @@ public class RetryTemplate implements RetryOperations {
|
||||
|
||||
doOnErrorInterceptors(retryCallback, context, e);
|
||||
|
||||
retryPolicy.registerThrowable(context, e);
|
||||
registerThrowable(retryPolicy, state, context, e);
|
||||
|
||||
if (shouldRethrow(context)) {
|
||||
if (shouldRethrow(retryPolicy, context, state)) {
|
||||
logger.debug("Rethrow in retry for policy: count=" + context.getRetryCount());
|
||||
throw e;
|
||||
}
|
||||
@@ -230,44 +271,139 @@ public class RetryTemplate implements RetryOperations {
|
||||
.getLastThrowable());
|
||||
}
|
||||
|
||||
return handleRetryExhausted(recoveryCallback, context);
|
||||
return handleRetryExhausted(recoveryCallback, context, state);
|
||||
|
||||
}
|
||||
finally {
|
||||
retryPolicy.close(context, lastException == null);
|
||||
close(retryPolicy, context, state, lastException == null);
|
||||
doCloseInterceptors(retryCallback, context, lastException);
|
||||
RetrySynchronizationManager.clear();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param context
|
||||
* @param state
|
||||
* @param succeeded
|
||||
*/
|
||||
protected void close(RetryPolicy retryPolicy, RetryContext context, RetryState state, boolean succeeded) {
|
||||
if (state != null) {
|
||||
if (succeeded) {
|
||||
retryContextCache.remove(state.getKey());
|
||||
retryPolicy.close(context, succeeded);
|
||||
}
|
||||
}
|
||||
else {
|
||||
retryPolicy.close(context, succeeded);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param retryPolicy
|
||||
* @param state
|
||||
* @param context
|
||||
* @param e
|
||||
*/
|
||||
protected void registerThrowable(RetryPolicy retryPolicy, RetryState state, RetryContext context, Exception e) {
|
||||
if (state != null) {
|
||||
Object key = state.getKey();
|
||||
// TODO: this comparison assumes that hashCode is the limiting
|
||||
// factor. Actually the cache should be able to decide for us.
|
||||
// if (initialHashCode != key.hashCode()) {
|
||||
// throw new RetryException(
|
||||
// "Inconsistent state for failed item key: hashCode has changed. "
|
||||
// +
|
||||
// "Consider whether equals() or hashCode() for the item might be inconsistent, "
|
||||
// + "or if you need to supply a better ItemKeyGenerator");
|
||||
// }
|
||||
retryContextCache.put(key, context);
|
||||
}
|
||||
retryPolicy.registerThrowable(context, e);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param retryCallback
|
||||
* @param retryPolicy
|
||||
* @return a retry context
|
||||
*/
|
||||
protected RetryContext open(RetryCallback retryCallback, RetryPolicy retryPolicy, RetryState state) {
|
||||
|
||||
// TODO: we don't need the callback here
|
||||
|
||||
if (state == null) {
|
||||
return doOpenInternal(retryCallback, retryPolicy);
|
||||
}
|
||||
|
||||
Object key = state.getKey();
|
||||
if (state.isForceRefresh()) {
|
||||
return doOpenInternal(retryCallback, retryPolicy);
|
||||
}
|
||||
else if (retryContextCache.containsKey(key)) {
|
||||
|
||||
RetryContext context = retryContextCache.get(key);
|
||||
if (context == null) {
|
||||
throw new RetryException("Inconsistent state for failed item: no history found. "
|
||||
+ "Consider whether equals() or hashCode() for the item might be inconsistent, "
|
||||
+ "or if you need to supply a better ItemKeyGenerator");
|
||||
}
|
||||
return context;
|
||||
|
||||
}
|
||||
else {
|
||||
|
||||
// The cache is only ued if there is a failure.
|
||||
return doOpenInternal(retryCallback, retryPolicy);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param retryCallback
|
||||
* @param retryPolicy
|
||||
* @return
|
||||
*/
|
||||
private RetryContext doOpenInternal(RetryCallback retryCallback, RetryPolicy retryPolicy) {
|
||||
return retryPolicy.open(retryCallback, RetrySynchronizationManager.getContext());
|
||||
}
|
||||
|
||||
/**
|
||||
* @param recoveryCallback the callback for recovery (might be null)
|
||||
* @param context the current retry context
|
||||
* @throws Exception if the callback does, and if there is no callback then
|
||||
* definitely the last exception from the context
|
||||
*/
|
||||
private Object handleRetryExhausted(RecoveryCallback recoveryCallback, RetryContext context) throws Exception {
|
||||
return retryPolicy.handleRetryExhausted(context);
|
||||
// if (recoveryCallback != null) {
|
||||
// return recoveryCallback.recover(context);
|
||||
// }
|
||||
// logger.debug("Retry exhausted after last attempt with no recovery path.");
|
||||
// throw context.getLastThrowable();
|
||||
protected Object handleRetryExhausted(RecoveryCallback recoveryCallback, RetryContext context, RetryState state)
|
||||
throws Exception {
|
||||
if (state != null) {
|
||||
retryContextCache.remove(state.getKey());
|
||||
}
|
||||
// TODO: test this when state==null
|
||||
if (recoveryCallback != null) {
|
||||
return recoveryCallback.recover(context);
|
||||
}
|
||||
if (state != null) {
|
||||
logger.debug("Retry exhausted after last attempt with no recovery path.");
|
||||
throw new ExhaustedRetryException("Retry exhausted after last attempt with no recovery path", context
|
||||
.getLastThrowable());
|
||||
}
|
||||
throw context.getLastThrowable();
|
||||
}
|
||||
|
||||
/**
|
||||
* Extension point for subclasses to decide on behaviour after catching an
|
||||
* exception in a {@link RetryCallback}. Normal stateless behaviour is not
|
||||
* to rethrow.
|
||||
* to rethrow, and if there is state we rethrow if the policy can still
|
||||
* retry.
|
||||
*
|
||||
* @param context the current {@link RetryContext}
|
||||
* @param retryPolicy
|
||||
* @param context the current context
|
||||
*
|
||||
* @return false but subclasses might choose otherwise
|
||||
*/
|
||||
protected boolean shouldRethrow(RetryContext context) {
|
||||
// TODO: return false
|
||||
return retryPolicy.shouldRethrow(context);
|
||||
protected boolean shouldRethrow(RetryPolicy retryPolicy, RetryContext context, RetryState state) {
|
||||
return state != null;
|
||||
}
|
||||
|
||||
private boolean doOpenInterceptors(RetryCallback callback, RetryContext context) {
|
||||
|
||||
@@ -1,58 +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.support;
|
||||
|
||||
import org.springframework.batch.retry.RecoveryCallback;
|
||||
import org.springframework.batch.retry.RetryCallback;
|
||||
import org.springframework.batch.retry.RetryContext;
|
||||
import org.springframework.batch.retry.RetryPolicy;
|
||||
|
||||
/**
|
||||
*
|
||||
*
|
||||
* @author Dave Syer
|
||||
*/
|
||||
public class StatefulRetryTemplate extends RetryTemplate {
|
||||
|
||||
/**
|
||||
* @param retryCallback
|
||||
* @param recoveryCallback
|
||||
* @param retryPolicy
|
||||
* @return the result of the callback
|
||||
* @throws Exception
|
||||
*/
|
||||
protected Object doExecute(RetryCallback retryCallback, RecoveryCallback recoveryCallback, RetryPolicy retryPolicy)
|
||||
throws Exception {
|
||||
|
||||
return super.doExecute(retryCallback, recoveryCallback, retryPolicy);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Extension point for subclasses to decide on behaviour after catching an
|
||||
* exception in a {@link RetryCallback}. Normal stateless behaviour is not
|
||||
* to rethrow.
|
||||
*
|
||||
* @param context the current {@link RetryContext}
|
||||
*
|
||||
* @return false but subclasses might choose otherwise
|
||||
*/
|
||||
protected boolean shouldRethrow(RetryContext context) {
|
||||
return false;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user