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

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

Tidied up test cases and removed DefaultStepExecutor (Simple* does everything).
This commit is contained in:
dsyer
2007-12-16 17:59:35 +00:00
parent 3c34d95ff8
commit c117a44354
9 changed files with 164 additions and 251 deletions

View File

@@ -1,88 +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.execution.step.simple;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.core.domain.StepInstance;
import org.springframework.batch.core.tasklet.Tasklet;
import org.springframework.batch.core.tasklet.Recoverable;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
/**
* Adds some recovery behaviour to {@link SimpleStepExecutor}.
*
* @author Dave Syer
*
*/
public class DefaultStepExecutor extends SimpleStepExecutor {
/**
* Extends {@link SimpleStepExecutor#doTaskletProcessing(Tasklet, StepInstance)} to
* add some basic recovery behaviour. If the {@link Tasklet} implements
* {@link Recoverable} <em>and</em> {@link Skippable} then the recovery
* and skip methods are called. The recovery is done in a new transaction,
* started with propagation
* {@link TransactionDefinition#PROPAGATION_REQUIRES_NEW} so that the
* inevitable rollback on the main processing loop does not cause the
* recovery to roll back as well.
*
* @throws Exception whenever {@link SimpleStepExecutor} would, but takes
* the recovery path first.
*
* @see org.springframework.batch.execution.step.simple.SimpleStepExecutor#doTaskletProcessing(org.springframework.batch.core.tasklet.Tasklet,
* org.springframework.batch.core.domain.StepInstance)
*/
protected ExitStatus doTaskletProcessing(Tasklet module, final StepExecution step) throws Exception {
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
try {
exitStatus = super.doTaskletProcessing(module, step);
}
catch (final Exception e) {
if (module instanceof Recoverable && module instanceof Skippable) {
final Recoverable recoverable = (Recoverable) module;
new TransactionTemplate(transactionManager, new DefaultTransactionDefinition(
TransactionDefinition.PROPAGATION_REQUIRES_NEW)).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
recoverable.recover(e);
return null;
}
});
}
if (module instanceof Skippable) {
((Skippable) module).skip();
}
// Rethrow so that outer transaction is rolled back properly
throw e;
}
return exitStatus;
}
}

View File

@@ -34,6 +34,7 @@ import org.springframework.batch.execution.scope.StepScope;
import org.springframework.batch.execution.scope.StepSynchronizationManager;
import org.springframework.batch.execution.step.RepeatOperationsHolder;
import org.springframework.batch.execution.step.SimpleStepConfiguration;
import org.springframework.batch.io.Skippable;
import org.springframework.batch.io.exception.BatchCriticalException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
@@ -357,7 +358,10 @@ public class SimpleStepExecutor implements StepExecutor {
/**
* Execute the business logic, delegating to the given {@link Tasklet}.
* Subclasses could extend the behaviour as long as they always return the
* value of this method call in their superclass.
* value of this method call in their superclass.<br/>
*
* If there is an exception and the {@link Tasklet} implements
* {@link Skippable} then the skip method is called.
*
* @param tasklet
* the unit of business logic to execute
@@ -369,7 +373,24 @@ public class SimpleStepExecutor implements StepExecutor {
*/
protected ExitStatus doTaskletProcessing(Tasklet tasklet,
StepExecution stepExecution) throws Exception {
return tasklet.execute();
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
try {
exitStatus = tasklet.execute();
} catch (Exception e) {
if (tasklet instanceof Skippable) {
((Skippable) tasklet).skip();
}
// Rethrow so that outer transaction is rolled back properly
throw e;
}
return exitStatus;
}
/**

View File

@@ -29,9 +29,6 @@ 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;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.callback.ItemProviderRetryCallback;
import org.springframework.batch.retry.policy.ItemProviderRetryPolicy;
@@ -61,11 +58,11 @@ import org.springframework.util.Assert;
* case because a transaction would have rolled back and the item would be
* represented).<br/>
*
* If neither a {@link RetryPolicy} nor a {@link RetryOperations} is provided
* then the {@link Recoverable} interface can be used to attempt to recover
* immediately (with no retry) from a processing error. Clients of this class
* must call {@link Recoverable#recover(Throwable)} directly, which is simply
* delegated to {@link ItemProvider#recover(Object, Throwable)}.
* If a {@link RetryPolicy} is not provided then the {@link Recoverable}
* interface can be used to attempt to recover immediately (with no retry) from
* a processing error. Clients of this class must call
* {@link Recoverable#recover(Throwable)} directly, which is simply delegated to
* {@link ItemProvider#recover(Object, Throwable)}.
*
* @see ItemProvider
* @see ItemProcessor
@@ -77,8 +74,8 @@ import org.springframework.util.Assert;
* @author Robert Kasanicky
*
*/
public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippable, StatisticsProvider,
InitializingBean {
public class ItemProviderProcessTasklet implements Tasklet, Skippable,
StatisticsProvider, InitializingBean {
/**
* Prefix added to statistics keys from processor if needed to avoid
@@ -92,93 +89,76 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
*/
public static final String PROVIDER_STATISTICS_PREFIX = "provider.";
/**
* Attribute key in the surrounding {@link RepeatContext} for the current
* item being processed. Needed to provide recoverable behavior if
* {@link RetryOperations} are not provided.
*/
private static final String ITEM_KEY = ItemProviderProcessTasklet.class.getName() + ".ITEM";
private RetryPolicy retryPolicy = null;
private RetryOperations retryOperations = null;
protected ItemProvider itemProvider;
protected ItemProcessor itemProcessor;
private ItemRecoverer itemRecoverer;
private RetryTemplate template = new RetryTemplate();
private ItemProviderRetryCallback callback;
/**
* Check mandatory properties (provider and processor), and ensure that only
* one (or neither) of {@link RetryPolicy} or {@link RetryOperations} is
* provided.
* Check mandatory properties (provider and processor).
*
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(itemProvider, "ItemProvider must be provided");
Assert.notNull(itemProcessor, "ItemProcessor must be provided");
Assert.state(!(retryPolicy != null && retryOperations != null),
"Either RetryOperations or RetryPolicy can be provided, but not both.");
if (retryPolicy != null) {
RetryTemplate template = new RetryTemplate();
template.setRetryPolicy(new ItemProviderRetryPolicy(retryPolicy));
retryOperations = template;
}
if (itemRecoverer==null && (itemProvider instanceof ItemRecoverer)) {
if (itemRecoverer == null && (itemProvider instanceof ItemRecoverer)) {
itemRecoverer = (ItemRecoverer) itemProvider;
}
ItemProviderRetryPolicy itemProviderRetryPolicy = new ItemProviderRetryPolicy(
retryPolicy);
template.setRetryPolicy(itemProviderRetryPolicy);
if (retryPolicy != null) {
callback = new ItemProviderRetryCallback(itemProvider,
itemProcessor);
callback.setRecoverer(itemRecoverer);
}
}
/**
* Read from the {@link ItemProvider} and process (if not null) with the
* {@link ItemProcessor}. The call to {@link ItemProcessor} is wrapped in a
* retry, if either a {@link RetryPolicy} or a {@link RetryOperations} is
* provided.
* stateful retry, if a {@link RetryPolicy} is provided. The
* {@link ItemRecoverer} is used (if provided) in the case of an exception
* to apply alternate processing to the item. If the stateful retry is in
* place then the recovery will happen in the next transaction
* automatically, otherwise it might be necessary for clients to make the
* recover method transactional with appropriate propagation behaviour
* (probably REQUIRES_NEW because the call will happen in the context of a
* transaction that is about to rollback).
*
* @see org.springframework.batch.core.tasklet.Tasklet#execute()
*/
public ExitStatus execute() throws Exception {
if (retryOperations != null) {
return new ExitStatus(retryOperations.execute(new ItemProviderRetryCallback(itemProvider, itemProcessor)) != null);
}
else {
Object data = itemProvider.next();
if (data == null) {
if (callback == null) {
Object item = itemProvider.next();
if (item == null) {
return ExitStatus.FINISHED;
}
RepeatContext context = RepeatSynchronizationManager.getContext();
Assert.state(context != null,
"No context available: you probably need to use this class inside a batch operation.");
context.setAttribute(ITEM_KEY, data);
itemProcessor.process(data);
// No exception so clear context (we can't recover directly because
// the current transaction is going to roll back)
context.removeAttribute(ITEM_KEY);
try {
itemProcessor.process(item);
} catch (Exception e) {
if (itemRecoverer != null) {
itemRecoverer.recover(item, e);
}
// Re-throw the exception so that the surrounding transaction
// rolls back if there is one
throw e;
}
return ExitStatus.CONTINUABLE;
}
}
/**
* Call out to the provider for recovery step.
*
* @see org.springframework.batch.core.tasklet.Recoverable#recover(java.lang.Throwable)
*/
public void recover(Throwable cause) {
RepeatContext context = RepeatSynchronizationManager.getContext();
Assert.state(context != null,
"No context available: you probably need to use this class inside a batch operation.");
try {
Object data = context.getAttribute(ITEM_KEY);
if (itemRecoverer!=null) {
itemRecoverer.recover(data, cause);
}
}
finally {
context.removeAttribute(ITEM_KEY);
}
return new ExitStatus(template.execute(callback) != null);
}
/**
@@ -200,7 +180,7 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
*
* @param itemRecoverer
*/
public void setRecoverer(ItemRecoverer itemRecoverer) {
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
this.itemRecoverer = itemRecoverer;
}
@@ -211,6 +191,11 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
* @see org.springframework.batch.io.Skippable#skip()
*/
public void skip() {
if (callback != null) {
// No need to skip because the recoverer will take any action
// necessary.
return;
}
if (this.itemProvider instanceof Skippable) {
((Skippable) this.itemProvider).skip();
}
@@ -235,9 +220,11 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
stats = ((StatisticsProvider) this.itemProvider).getStatistics();
}
if (this.itemProcessor instanceof StatisticsProvider) {
Properties props = ((StatisticsProvider) this.itemProcessor).getStatistics();
Properties props = ((StatisticsProvider) this.itemProcessor)
.getStatistics();
if (!stats.isEmpty()) {
stats = prependKeys(stats, props, PROVIDER_STATISTICS_PREFIX, PROCESSOR_STATISTICS_PREFIX);
stats = prependKeys(stats, props, PROVIDER_STATISTICS_PREFIX,
PROCESSOR_STATISTICS_PREFIX);
} else {
stats.putAll(props);
}
@@ -250,10 +237,12 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
* @param string
* @return
*/
private Properties prependKeys(Properties props1, Properties props2, String prefix1, String prefix2) {
private Properties prependKeys(Properties props1, Properties props2,
String prefix1, String prefix2) {
Properties result = new Properties();
Set duplicates = new HashSet();
for (Iterator iterator = props1.entrySet().iterator(); iterator.hasNext();) {
for (Iterator iterator = props1.entrySet().iterator(); iterator
.hasNext();) {
Map.Entry entry = (Map.Entry) iterator.next();
String key = (String) entry.getKey();
String value = (String) entry.getValue();
@@ -263,7 +252,8 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
}
result.setProperty(key, value);
}
for (Iterator iterator = props2.entrySet().iterator(); iterator.hasNext();) {
for (Iterator iterator = props2.entrySet().iterator(); iterator
.hasNext();) {
Map.Entry entry = (Map.Entry) iterator.next();
String key = (String) entry.getKey();
String value = (String) entry.getValue();
@@ -274,8 +264,8 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
}
for (Iterator iterator = duplicates.iterator(); iterator.hasNext();) {
String key = (String) iterator.next();
result.setProperty(prefix1+key, props1.getProperty(key));
result.setProperty(prefix2+key, props2.getProperty(key));
result.setProperty(prefix1 + key, props1.getProperty(key));
result.setProperty(prefix2 + key, props2.getProperty(key));
}
return result;
}
@@ -283,18 +273,10 @@ public class ItemProviderProcessTasklet implements Tasklet, Recoverable, Skippab
/**
* Public setter for the retryPolicy.
*
* @param retyPolicy the retryPolicy to set
* @param retyPolicy
* the retryPolicy to set
*/
public void setRetryPolicy(RetryPolicy retryPolicy) {
this.retryPolicy = retryPolicy;
}
/**
* Public setter for the retryOperations.
*
* @param retryOperations the retryOperations to set
*/
public void setRetryOperations(RetryOperations retryOperations) {
this.retryOperations = retryOperations;
}
}

View File

@@ -152,7 +152,7 @@ public class DefaultJobExecutorTests extends TestCase {
checkRepository(BatchStatus.COMPLETED);
}
public void testRunWithDefaultStepExecutor() throws Exception {
public void testRunWithSimpleStepExecutor() throws Exception {
jobExecutor = new DefaultJobExecutor();
jobExecutor.setJobRepository(jobRepository);

View File

@@ -38,10 +38,11 @@ import org.springframework.batch.execution.repository.dao.MapJobDao;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.execution.runtime.ScheduledJobIdentifierFactory;
import org.springframework.batch.execution.step.SimpleStepConfiguration;
import org.springframework.batch.execution.step.simple.DefaultStepExecutor;
import org.springframework.batch.execution.step.simple.SimpleStepExecutor;
import org.springframework.batch.execution.tasklet.ItemProviderProcessTasklet;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemProvider;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.provider.ListItemProvider;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
@@ -54,9 +55,7 @@ import org.springframework.util.StringUtils;
public class SimpleJobTests extends TestCase {
private List list = new ArrayList();
// private int count;
private List recovered = new ArrayList();
private SimpleJobRepository repository = new SimpleJobRepository(new MapJobDao(), new MapStepDao());
@@ -72,7 +71,7 @@ public class SimpleJobTests extends TestCase {
private DefaultJobExecutor jobExecutor = new DefaultJobExecutor();;
private DefaultStepExecutor stepLifecycle = new DefaultStepExecutor();
private SimpleStepExecutor stepLifecycle = new SimpleStepExecutor();
protected void setUp() throws Exception {
super.setUp();
@@ -93,17 +92,18 @@ public class SimpleJobTests extends TestCase {
return getTasklet(new String[] { arg0, arg1 });
}
private Tasklet getTasklet(String[] args) throws Exception {
private ItemProviderProcessTasklet getTasklet(String[] args) throws Exception {
ItemProviderProcessTasklet module = new ItemProviderProcessTasklet();
List items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(args));
provider = new ListItemProvider(items) {
provider = new ListItemProvider(items);
module.setItemRecoverer(new ItemRecoverer() {
public boolean recover(Object item, Throwable cause) {
list.add(item);
recovered.add(item);
assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
return true;
}
};
});
module.setItemProvider(provider);
module.setItemProcessor(processor);
module.afterPropertiesSet();
@@ -161,37 +161,38 @@ public class SimpleJobTests extends TestCase {
* is recovered ("skipped") on the second attempt (see retry policy
* definition above)...
*/
final Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
final ItemProviderProcessTasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
StepConfiguration step = new SimpleStepConfiguration(module);
((ItemProviderProcessTasklet) module).setItemProcessor(new ItemProcessor() {
module.setItemProcessor(new ItemProcessor() {
public void process(Object data) throws Exception {
throw new RuntimeException("Try again Dummy!");
}
});
module.afterPropertiesSet();
jobConfiguration.addStep(step);
JobExecution jobExecution = repository.findOrCreateJob(jobConfiguration, runtimeInformation);
JobInstance job = jobExecution.getJob();
jobExecutor.run(jobConfiguration, jobExecution);
assertEquals(BatchStatus.COMPLETED, job.getStatus());
assertEquals(BatchStatus.COMPLETED, jobExecution.getJob().getStatus());
assertEquals(0, processed.size());
// provider should be exhausted
assertEquals(null, provider.next());
assertEquals(3, list.size());
assertEquals(3, recovered.size());
}
public void testExceptionTerminates() throws Exception {
JobConfiguration jobConfiguration = new JobConfiguration();
JobIdentifier runtimeInformation = new SimpleJobIdentifier("real.job");
final Tasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
final ItemProviderProcessTasklet module = getTasklet(new String[] { "foo", "bar", "spam" });
StepConfiguration step = new SimpleStepConfiguration(module);
((ItemProviderProcessTasklet) module).setItemProcessor(new ItemProcessor() {
module.setItemProcessor(new ItemProcessor() {
public void process(Object data) throws Exception {
throw new RuntimeException("Foo");
}
});
module.afterPropertiesSet();
jobConfiguration.addStep(step);
JobExecution jobExecution = repository.findOrCreateJob(jobConfiguration, runtimeInformation);

View File

@@ -60,7 +60,7 @@ public class DefaultStepExecutorTests extends TestCase {
}
};
private DefaultStepExecutor stepExecutor;
private SimpleStepExecutor stepExecutor;
private StepConfigurationSupport stepConfiguration;
@@ -73,11 +73,13 @@ public class DefaultStepExecutorTests extends TestCase {
/**
* @param strings
* @return
* @throws Exception
*/
private Tasklet getTasklet(String[] strings) {
private Tasklet getTasklet(String[] strings) throws Exception {
ItemProviderProcessTasklet module = new ItemProviderProcessTasklet();
module.setItemProcessor(processor);
module.setItemProvider(getProvider(strings));
module.afterPropertiesSet();
return module;
}
@@ -88,7 +90,7 @@ public class DefaultStepExecutorTests extends TestCase {
*/
protected void setUp() throws Exception {
super.setUp();
stepExecutor = new DefaultStepExecutor();
stepExecutor = new SimpleStepExecutor();
stepExecutor.setRepository(new JobRepositorySupport());
stepConfiguration = new SimpleStepConfiguration();
stepConfiguration.setTasklet(getTasklet(new String[] { "foo", "bar",

View File

@@ -23,15 +23,14 @@ import java.util.Properties;
import junit.framework.TestCase;
import org.springframework.batch.execution.tasklet.ItemProviderProcessTasklet;
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.item.provider.AbstractItemProvider;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.batch.repeat.synch.RepeatSynchronizationManager;
import org.springframework.batch.retry.policy.NeverRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.statistics.StatisticsProvider;
import org.springframework.batch.support.PropertiesConverter;
@@ -68,7 +67,7 @@ public class ItemProviderProcessTaskletTests extends TestCase {
private ItemProviderProcessTasklet module;
public void setUp() {
public void setUp() throws Exception {
// create module
module = new ItemProviderProcessTasklet();
@@ -77,12 +76,15 @@ public class ItemProviderProcessTaskletTests extends TestCase {
module.setItemProvider(itemProvider);
module.setItemProcessor(itemProcessor);
module.afterPropertiesSet();
RepeatSynchronizationManager.register(new RepeatContextSupport(null));
}
/*
* (non-Javadoc)
*
* @see junit.framework.TestCase#tearDown()
*/
protected void tearDown() throws Exception {
@@ -128,8 +130,7 @@ public class ItemProviderProcessTaskletTests extends TestCase {
module.execute();
// TODO: should we expect Batch exception?
fail("RuntimeException was expected");
}
catch (RuntimeException bce) {
} catch (RuntimeException bce) {
// expected
}
}
@@ -145,6 +146,7 @@ public class ItemProviderProcessTaskletTests extends TestCase {
public void testSkippableProvider() throws Exception {
module.setItemProvider(new SkippableItemProvider());
module.setItemRecoverer(null);
module.skip();
assertEquals(1, list.size());
}
@@ -152,17 +154,18 @@ public class ItemProviderProcessTaskletTests extends TestCase {
public void testSkippablProviderProcessor() throws Exception {
module.setItemProvider(new SkippableItemProvider());
module.setItemProcessor(new SkippableItemProcessor());
module.setItemRecoverer(null);
module.skip();
assertEquals(2, list.size());
}
public void testStatisticsProvider() throws Exception {
module.setItemProvider(new SkippableItemProvider());
Properties stats = module.getStatistics();
assertEquals(1, stats.size());
assertEquals("bar", stats.getProperty("foo"));
}
public void testStatisticsProcessor() throws Exception {
module.setItemProcessor(new SkippableItemProcessor());
Properties stats = module.getStatistics();
@@ -179,9 +182,11 @@ public class ItemProviderProcessTaskletTests extends TestCase {
assertEquals("bar", stats.getProperty("processor.foo"));
}
public void testStatisticsProviderProcessorMergeDuplicates() throws Exception {
public void testStatisticsProviderProcessorMergeDuplicates()
throws Exception {
module.setItemProvider(new SkippableItemProvider());
module.setItemProcessor(new SkippableItemProcessor("foo=bar\nspam=bucket"));
module.setItemProcessor(new SkippableItemProcessor(
"foo=bar\nspam=bucket"));
Properties stats = module.getStatistics();
assertEquals(3, stats.size());
assertEquals("bar", stats.getProperty("provider.foo"));
@@ -194,17 +199,20 @@ public class ItemProviderProcessTaskletTests extends TestCase {
// set up and call execute
items = Collections.singletonList("foo");
module.setItemProvider(new AbstractItemProvider() {
module.setItemRecoverer(new ItemRecoverer() {
public boolean recover(Object item, Throwable cause) {
assertEquals("foo", cause.getMessage());
assertEquals("FOO", cause.getMessage());
list.add(item);
return true;
}
});
module.setItemProvider(new AbstractItemProvider() {
public Object next() throws Exception {
return itemProvider.next();
return "bar";
}
});
module.setItemProcessor(new ItemProcessor() {
public void process(Object data) throws Exception {
throw new RuntimeException("FOO");
@@ -212,44 +220,34 @@ public class ItemProviderProcessTaskletTests extends TestCase {
});
module.afterPropertiesSet();
try {
module.execute();
fail("Expected RuntimeException");
}
catch (RuntimeException e) {
} catch (RuntimeException e) {
assertEquals("FOO", e.getMessage());
}
list.clear();
// After a processing exception client has to call recover directly
module.recover(new RuntimeException("foo"));
// After a processing exception the recovery is done automatically.
// verify method calls
assertEquals(1, list.size());
assertEquals("The item was not passed in to recover method", "foo", list.get(0));
assertEquals("The item was not passed in to recover method", "bar",
list.get(0));
}
public void testRetryPolicy() throws Exception {
module.setRetryPolicy(new NeverRetryPolicy());
// set up mock objects
items = new ArrayList() {
{
add("foo");
add("foo"); // in production use this would be the second
// attempt after rollback
}
};
module.setItemProvider(new AbstractItemProvider() {
module.setRetryPolicy(new SimpleRetryPolicy(1));
module.setItemRecoverer(new ItemRecoverer() {
public boolean recover(Object item, Throwable cause) {
assertEquals("FOO", cause.getMessage());
list.add(item + "_recovered");
list.add(item+"_recovered");
return true;
}
});
module.setItemProvider(new AbstractItemProvider() {
public Object next() throws Exception {
return itemProvider.next();
return "foo";
}
});
module.setItemProcessor(new ItemProcessor() {
@@ -264,8 +262,7 @@ public class ItemProviderProcessTaskletTests extends TestCase {
try {
module.execute();
fail("Expected RuntimeException");
}
catch (RuntimeException e) {
} catch (RuntimeException e) {
assertEquals("FOO", e.getMessage());
}
@@ -276,15 +273,15 @@ public class ItemProviderProcessTaskletTests extends TestCase {
// verify method calls
assertEquals(1, list.size());
assertEquals("The item was not passed in to recover method", "foo_recovered", list.get(0));
assertEquals("The item was not passed in to recover method",
"foo_recovered", list.get(0));
}
public void testInitialisationWithNullProvider() throws Exception {
module.setItemProvider(null);
try {
module.afterPropertiesSet();
}
catch (IllegalArgumentException e) {
} catch (IllegalArgumentException e) {
assertTrue(e.getMessage().toLowerCase().indexOf("provider") >= 0);
}
}
@@ -293,27 +290,19 @@ public class ItemProviderProcessTaskletTests extends TestCase {
module.setItemProcessor(null);
try {
module.afterPropertiesSet();
}
catch (IllegalArgumentException e) {
} catch (IllegalArgumentException e) {
assertTrue(e.getMessage().toLowerCase().indexOf("processor") >= 0);
}
}
public void testInitialisationWithNotNullPolicyAndOperations() throws Exception {
module.setRetryPolicy(new NeverRetryPolicy());
module.setRetryOperations(new RetryTemplate());
try {
module.afterPropertiesSet();
}
catch (IllegalStateException e) {
assertTrue(e.getMessage().toLowerCase().indexOf("not both") >= 0);
}
}
private class SkippableItemProvider extends AbstractItemProvider implements Skippable, StatisticsProvider {
private class SkippableItemProvider implements ItemProvider,
Skippable, StatisticsProvider {
public Object next() throws Exception {
return itemProvider.next();
}
public Object getKey(Object item) {
return item;
}
public void skip() {
list.add("provider");
}
@@ -322,21 +311,27 @@ public class ItemProviderProcessTaskletTests extends TestCase {
}
}
private class SkippableItemProcessor implements ItemProcessor, Skippable, StatisticsProvider {
private class SkippableItemProcessor implements ItemProcessor, Skippable,
StatisticsProvider {
String props = "foo=bar";
public SkippableItemProcessor() {
super();
}
public SkippableItemProcessor(String props) {
this();
this.props = props;
}
public void process(Object data) throws Exception {
// no-op
}
public void skip() {
list.add("processor");
}
public Properties getStatistics() {
return PropertiesConverter.stringToProperties(props);
}

View File

@@ -56,7 +56,7 @@
</bean>
<bean id="stepExecutor"
class="org.springframework.batch.execution.step.simple.DefaultStepExecutor"
class="org.springframework.batch.execution.step.simple.SimpleStepExecutor"
scope="prototype">
<property name="transactionManager" ref="transactionManager" />
<property name="repository" ref="simpleJobRepository" />

View File

@@ -60,7 +60,7 @@ public class HibernateFailureJobFunctionalTests extends
throw e;
} catch (UncategorizedSQLException e) {
// This is what would happen if the job wasn't configured to skip
// exceptions at teh step level.
// exceptions at the step level.
assertEquals(1, writer.getErrors().size());
throw e;
}