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

OPEN - issue BATCH-87: Recoverable is a nasty abstraction - encourages stateful Tasklets.

Browse Source 
http://opensource.atlassian.com/projects/spring/browse/BATCH-87

Moved recovery responsibility to a separate infrastructure interface.
This commit is contained in:
dsyer
2007-12-16 14:22:34 +00:00
parent a48b7847e3
commit 3c34d95ff8
11 changed files with 187 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemProvider.java
View File

@@ -48,18 +48,6 @@ public interface ItemProvider {
*/
Object next() throws Exception;
/**
* Recover gracefully from an error. Clients can call this if processing of
* the item throws an unexpected exception. Caller can use the return value
* to decide whether to try more corrective action or perhaps throw an
* exception.
*
* @param data the item that failed.
* @param cause the cause of the failure that led to this recovery.
* @return true if recovery was successful.
*/
boolean recover(Object data, Throwable cause);
/**
* Get a unique identifier for the item that can be used to cache it between
* calls if necessary, and then identify it later.

39
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* 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.item;
/**
* Strategy interface for recovery action when processing of an item fails.<br/>
*
* @author Dave Syer
*/
public interface ItemRecoverer {
/**
* Recover gracefully from an error. Clients can call this if processing of
* the item throws an unexpected exception. Caller can use the return value
* to decide whether to try more corrective action or perhaps throw an
* exception.
*
* @param data
* the item that failed.
* @param cause
* the cause of the failure that led to this recovery.
* @return true if recovery was successful.
*/
boolean recover(Object data, Throwable cause);
}

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/provider/AbstractItemProvider.java
View File

@@ -17,8 +17,9 @@
package org.springframework.batch.item.provider;
import org.springframework.batch.item.ItemProvider;
import org.springframework.batch.item.ItemRecoverer;
public abstract class AbstractItemProvider implements ItemProvider {
public abstract class AbstractItemProvider implements ItemProvider, ItemRecoverer {
/**
* Do nothing. Subclassses should override to implement recovery behaviour.

49
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemProviderRetryCallback.java
View File

@@ -20,6 +20,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemProvider;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.RetryPolicy;
@@ -40,20 +41,34 @@ import org.springframework.batch.retry.policy.ItemProviderRetryPolicy;
*/
public class ItemProviderRetryCallback implements RetryCallback {
private final static Log logger = LogFactory.getLog(ItemProviderRetryCallback.class);
private final static Log logger = LogFactory
.getLog(ItemProviderRetryCallback.class);
public static final String ITEM = ItemProviderRetryCallback.class.getName() + ".ITEM";
public static final String ITEM = ItemProviderRetryCallback.class.getName()
+ ".ITEM";
private ItemProvider provider;
private ItemProcessor processor;
public ItemProviderRetryCallback(ItemProvider provider, ItemProcessor processor) {
private ItemRecoverer recoverer;
public ItemProviderRetryCallback(ItemProvider provider,
ItemProcessor processor) {
super();
this.provider = provider;
this.processor = processor;
}
/**
* Setter for injecting optional recovery handler.
*
* @param recoveryHandler
*/
public void setRecoverer(ItemRecoverer recoverer) {
this.recoverer = recoverer;
}
public Object doWithRetry(RetryContext context) throws Throwable {
// This requires a collaboration with the RetryPolicy...
if (!context.isExhaustedOnly()) {
@@ -67,9 +82,9 @@ public class ItemProviderRetryCallback implements RetryCallback {
if (item == null) {
try {
item = provider.next();
}
catch (Exception e) {
throw new ExhaustedRetryException("Unexpected end of item provider", e);
} catch (Exception e) {
throw new ExhaustedRetryException(
"Unexpected end of item provider", e);
}
if (item == null) {
// This is probably not fatal: in a batch we want to
@@ -91,8 +106,26 @@ public class ItemProviderRetryCallback implements RetryCallback {
}
/**
* Accessor for the {@link ItemProvider}.
* @return the provider.
* Accessor for the {@link ItemRecoverer}. If the handler is null but
* the {@link ItemProvider} is an instanceof {@link ItemRecoverer},
* then it will be returned instead.
*
* @return the {@link ItemRecoverer}.
*/
public ItemRecoverer getRecoverer() {
if (recoverer != null) {
return recoverer;
}
if (provider instanceof ItemRecoverer) {
return (ItemRecoverer) provider;
}
return null;
}
/**
* Public getter for the {@link ItemProvider}.
*
* @return the {@link ItemProvider} instance.
*/
public ItemProvider getProvider() {
return provider;

60
spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemProviderRetryPolicy.java
View File

@@ -20,6 +20,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.FailedItemIdentifier;
import org.springframework.batch.item.ItemProvider;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
@@ -44,7 +45,9 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
protected Log logger = LogFactory.getLog(getClass());
public static final String EXHAUSTED = ItemProviderRetryPolicy.class.getName() + ".EXHAUSTED";
public static final String EXHAUSTED = ItemProviderRetryPolicy.class
.getName()
+ ".EXHAUSTED";
private RetryPolicy delegate;
@@ -100,12 +103,14 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
*
* @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback)
*
* @throws IllegalStateException if the callback is not of the required
* type.
* @throws IllegalStateException
* if the callback is not of the required type.
*/
public RetryContext open(RetryCallback callback) {
Assert.state(callback instanceof ItemProviderRetryCallback, "Callback must be ItemProviderRetryCallback");
ItemProviderRetryContext context = new ItemProviderRetryContext((ItemProviderRetryCallback) callback);
Assert.state(callback instanceof ItemProviderRetryCallback,
"Callback must be ItemProviderRetryCallback");
ItemProviderRetryContext context = new ItemProviderRetryContext(
(ItemProviderRetryCallback) callback);
context.open(callback);
return context;
}
@@ -115,9 +120,10 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
* implemented by subclasses), and remove the current item from the history.
*
* @see org.springframework.batch.retry.RetryPolicy#registerThrowable(org.springframework.batch.retry.RetryContext,
* java.lang.Throwable)
* java.lang.Throwable)
*/
public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException {
public void registerThrowable(RetryContext context, Throwable throwable)
throws TerminatedRetryException {
((RetryPolicy) context).registerThrowable(context, throwable);
// The throwable is stored in the delegate context.
}
@@ -131,7 +137,8 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
return ((RetryPolicy) context).handleRetryExhausted(context);
}
private class ItemProviderRetryContext extends RetryContextSupport implements RetryPolicy {
private class ItemProviderRetryContext extends RetryContextSupport
implements RetryPolicy {
private Object item;
@@ -140,10 +147,13 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
private ItemProvider provider;
private ItemRecoverer recoverer;
public ItemProviderRetryContext(ItemProviderRetryCallback callback) {
super(RetrySynchronizationManager.getContext());
item = callback.next(this);
this.provider = callback.getProvider();
this.recoverer = callback.getRecoverer();
}
public boolean canRetry(RetryContext context) {
@@ -156,7 +166,8 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
public RetryContext open(RetryCallback callback) {
if (hasFailed(provider, item)) {
this.delegateContext = retryContextCache.get(provider.getKey(item));
this.delegateContext = retryContextCache.get(provider
.getKey(item));
}
if (this.delegateContext == null) {
// Only create a new context if we don't know the history of
@@ -167,31 +178,42 @@ public class ItemProviderRetryPolicy extends AbstractStatefulRetryPolicy {
return null;
}
public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException {
public void registerThrowable(RetryContext context, Throwable throwable)
throws TerminatedRetryException {
retryContextCache.put(provider.getKey(item), this.delegateContext);
delegate.registerThrowable(this.delegateContext, throwable);
}
public boolean isExternal() {
// Not called...
throw new UnsupportedOperationException("Not supported - this code should be unreachable.");
throw new UnsupportedOperationException(
"Not supported - this code should be unreachable.");
}
public boolean shouldRethrow(RetryContext context) {
// Not called...
throw new UnsupportedOperationException("Not supported - this code should be unreachable.");
throw new UnsupportedOperationException(
"Not supported - this code should be unreachable.");
}
public Object handleRetryExhausted(RetryContext context) throws Exception {
public Object handleRetryExhausted(RetryContext context)
throws Exception {
// If there is no going back, then we can remove the history
retryContextCache.remove(provider.getKey(item));
RepeatSynchronizationManager.setCompleteOnly();
boolean success = provider.recover(item, context.getLastThrowable());
if (!success) {
//TODO if context was null, there would be exception while getting success value
String count = context != null ? "" + context.getRetryCount() : "unknown";
logger.error("Could not recover from error after retry exhausted after [" + count + "] attempts.",
context.getLastThrowable());
if (recoverer != null) {
boolean success = recoverer.recover(item, context
.getLastThrowable());
if (!success) {
// TODO if context was null, there would be exception while
// getting success value
String count = context != null ? ""
+ context.getRetryCount() : "unknown";
logger.error(
"Could not recover from error after retry exhausted after ["
+ count + "] attempts.", context
.getLastThrowable());
}
}
return item;
}