diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTasklet.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTasklet.java index 31f4984bb..a1fc5bfa9 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTasklet.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTasklet.java @@ -27,6 +27,7 @@ import org.springframework.batch.core.tasklet.Tasklet; import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemProvider; +import org.springframework.batch.item.ItemRecoverer; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; @@ -106,6 +107,8 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab protected ItemProcessor itemProcessor; + private ItemRecoverer itemRecoverer; + /** * Check mandatory properties (provider and processor), and ensure that only * one (or neither) of {@link RetryPolicy} or {@link RetryOperations} is @@ -123,6 +126,9 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab template.setRetryPolicy(new ItemProviderRetryPolicy(retryPolicy)); retryOperations = template; } + if (itemRecoverer==null && (itemProvider instanceof ItemRecoverer)) { + itemRecoverer = (ItemRecoverer) itemProvider; + } } /** @@ -166,7 +172,9 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab try { Object data = context.getAttribute(ITEM_KEY); - itemProvider.recover(data, cause); + if (itemRecoverer!=null) { + itemRecoverer.recover(data, cause); + } } finally { context.removeAttribute(ITEM_KEY); @@ -187,6 +195,15 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab this.itemProcessor = moduleProcessor; } + /** + * Setter for injecting optional recovery handler. + * + * @param itemRecoverer + */ + public void setRecoverer(ItemRecoverer itemRecoverer) { + this.itemRecoverer = itemRecoverer; + } + /** * If the provider and / or processor are {@link Skippable} then delegate to * them in that order. diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java index dd9360e6f..de3aa9ea5 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/launch/SimpleJobTests.java @@ -85,15 +85,15 @@ public class SimpleJobTests extends TestCase { }); } - private Tasklet getTasklet(String arg) { + private Tasklet getTasklet(String arg) throws Exception { return getTasklet(new String[] { arg }); } - private Tasklet getTasklet(String arg0, String arg1) { + private Tasklet getTasklet(String arg0, String arg1) throws Exception { return getTasklet(new String[] { arg0, arg1 }); } - private Tasklet getTasklet(String[] args) { + private Tasklet getTasklet(String[] args) throws Exception { ItemProviderProcessTasklet module = new ItemProviderProcessTasklet(); List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(args)); @@ -106,6 +106,7 @@ public class SimpleJobTests extends TestCase { }; module.setItemProvider(provider); module.setItemProcessor(processor); + module.afterPropertiesSet(); return module; } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTaskletTests.java index fd542e816..fb5effaa5 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTaskletTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemProviderProcessTaskletTests.java @@ -211,6 +211,8 @@ public class ItemProviderProcessTaskletTests extends TestCase { } }); + module.afterPropertiesSet(); + try { module.execute(); fail("Expected RuntimeException"); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemProvider.java index a9403c569..25f9a0c4d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemProvider.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemProvider.java @@ -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. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java new file mode 100644 index 000000000..4f3e54f50 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemRecoverer.java @@ -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.
+ * + * @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); +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/provider/AbstractItemProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/provider/AbstractItemProvider.java index 7ca6cc20a..01b87eb0d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/provider/AbstractItemProvider.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/provider/AbstractItemProvider.java @@ -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. diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemProviderRetryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemProviderRetryCallback.java index 3ef27a70c..5cd28053c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemProviderRetryCallback.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemProviderRetryCallback.java @@ -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; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemProviderRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemProviderRetryPolicy.java index c2e8e487f..fb734600c 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemProviderRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemProviderRetryPolicy.java @@ -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; } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemProviderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemProviderTests.java index a7c4736f3..5ad83f0d1 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemProviderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemProviderTests.java @@ -32,12 +32,4 @@ public class ItemProviderTests extends TestCase { assertEquals("foo", provider.next()); } - public void testRecover() throws Exception { - try { - provider.recover("foo", null); - } - catch (Exception e) { - fail("Unexpected Exception"); - } - } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java new file mode 100644 index 000000000..eaa558cbc --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/ItemRecoveryHandlerTests.java @@ -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; + +import junit.framework.TestCase; + +import org.springframework.batch.item.provider.AbstractItemProvider; + +public class ItemRecoveryHandlerTests extends TestCase { + + ItemRecoverer recoverer = new AbstractItemProvider() { + public Object next() { + return "foo"; + } + }; + + public void testRecover() throws Exception { + try { + recoverer.recover("foo", null); + } + catch (Exception e) { + fail("Unexpected Exception"); + } + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemProviderRetryCallbackTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemProviderRetryCallbackTests.java index f2b88ad06..82d321da7 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemProviderRetryCallbackTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemProviderRetryCallbackTests.java @@ -161,7 +161,7 @@ public class ItemProviderRetryCallbackTests extends TestCase { } public void testRecoverWithoutSession() throws Exception { - callback.getProvider().recover("foo", null); + callback.getRecoverer().recover("foo", null); assertEquals(1, count); assertEquals(1, calls.size()); }