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

OPEN - issue BATCH-486: Users must choose between skip and retry

Browse Source 
Removed itemReader from retry callback - simplifies stateful retry and makes the ugly exception handling problem go away.
This commit is contained in:
dsyer
2008-03-27 15:21:20 +00:00
parent d983eba1c5
commit 0f84d3064a
10 changed files with 128 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,17 +17,17 @@ package org.springframework.batch.core.step.item;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.item.AbstractItemWriter;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.retry.RetryListener;
import org.springframework.batch.retry.RetryOperations;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.backoff.BackOffPolicy;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.policy.ItemWriterRetryPolicy;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
@@ -139,10 +139,7 @@ public class StatefulRetryStepFactoryBean extends SimpleStepFactoryBean {
getStepOperations()
.setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler()));
ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), itemKeyGenerator,
getItemWriter());
retryCallback.setRecoverer(itemRecoverer);
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
ItemWriterRetryPolicy itemProviderRetryPolicy = new ItemWriterRetryPolicy(retryPolicy);
RetryTemplate retryTemplate = new RetryTemplate();
if (retryListeners!=null) {
@@ -154,7 +151,7 @@ public class StatefulRetryStepFactoryBean extends SimpleStepFactoryBean {
}
StatefulRetryItemHandler itemHandler = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
retryTemplate, retryCallback);
retryTemplate, itemKeyGenerator, itemRecoverer);
step.setItemHandler(itemHandler);
@@ -166,21 +163,25 @@ public class StatefulRetryStepFactoryBean extends SimpleStepFactoryBean {
final private RetryOperations retryOperations;
final private ItemReaderRetryCallback retryCallback;
final private ItemKeyGenerator itemKeyGenerator;
final private ItemRecoverer itemRecoverer;
/**
* @param itemReader
* @param itemWriter
* @param retryCallback
* @param retryTemplate
* @param itemRecoverer
*/
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate,
ItemReaderRetryCallback retryCallback) {
ItemKeyGenerator itemKeyGenerator, ItemRecoverer itemRecoverer) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.retryCallback = retryCallback;
this.itemKeyGenerator = itemKeyGenerator;
this.itemRecoverer = itemRecoverer;
}
/**
* Execute the business logic, delegating to the reader and writer.
* Subclasses could extend the behaviour as long as they always return
@@ -197,13 +198,17 @@ public class StatefulRetryStepFactoryBean extends SimpleStepFactoryBean {
* call will happen in the context of a transaction that is about to
* rollback).<br/>
*
* @param contribution the current step
* @return {@link ExitStatus#CONTINUABLE} if there is more processing to
* do
* @throws Exception if there is an error
* @see org.springframework.batch.core.step.item.SimpleItemHandler#write(java.lang.Object, org.springframework.batch.core.StepContribution)
*/
public ExitStatus handle(StepContribution contribution) throws Exception {
return new ExitStatus(retryOperations.execute(retryCallback) != null);
protected void write(Object item, final StepContribution contribution) throws Exception {
ItemWriterRetryCallback retryCallback = new ItemWriterRetryCallback(item, new AbstractItemWriter() {
public void write(Object item) throws Exception {
doWrite(item);
}
});
retryCallback.setKeyGenerator(itemKeyGenerator);
retryCallback.setRecoverer(itemRecoverer);
retryOperations.execute(retryCallback);
}
}

3
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/validator/ValidatingItemReader.java
View File

@@ -15,9 +15,6 @@
*/
package org.springframework.batch.item.validator;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.support.DelegatingItemReader;
import org.springframework.util.Assert;

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

@@ -16,38 +16,34 @@
package org.springframework.batch.retry.callback;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.FailedItemIdentifier;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.retry.ExhaustedRetryException;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.RetryException;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.policy.ItemWriterRetryPolicy;
/**
* A {@link RetryCallback} that knows about and caches the value from an
* {@link ItemReader}. Used by the {@link ItemReaderRetryPolicy} to enable
* external retry of the item processing.
* A {@link RetryCallback} that knows about and caches an item, and attempts to
* process it using an {@link ItemWriter}. Used by the
* {@link ItemWriterRetryPolicy} to enable external retry of the item
* processing.
*
* @author Dave Syer
*
* @see ItemReaderRetryPolicy
* @see ItemWriterRetryPolicy
* @see RetryPolicy#handleRetryExhausted(RetryContext)
*
*/
public class ItemReaderRetryCallback implements RetryCallback {
public class ItemWriterRetryCallback implements RetryCallback {
private final static Log logger = LogFactory.getLog(ItemReaderRetryCallback.class);
public static final String ITEM = ItemWriterRetryCallback.class.getName() + ".ITEM";
public static final String ITEM = ItemReaderRetryCallback.class.getName() + ".ITEM";
private ItemReader reader;
private Object item;
private ItemWriter writer;
@@ -63,13 +59,13 @@ public class ItemReaderRetryCallback implements RetryCallback {
}
};
public ItemReaderRetryCallback(ItemReader reader, ItemWriter writer) {
this(reader, null, writer);
public ItemWriterRetryCallback(Object item, ItemWriter writer) {
this(item, writer, null);
}
public ItemReaderRetryCallback(ItemReader reader, ItemKeyGenerator keyGenerator, ItemWriter writer) {
public ItemWriterRetryCallback(Object item, ItemWriter writer, ItemKeyGenerator keyGenerator) {
super();
this.reader = reader;
this.item = item;
this.writer = writer;
this.keyGenerator = keyGenerator;
}
@@ -119,17 +115,7 @@ public class ItemReaderRetryCallback implements RetryCallback {
public Object next(RetryContext context) {
Object item = context.getAttribute(ITEM);
if (item == null) {
try {
item = reader.read();
}
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
// exit gracefully...
logger.info("ItemProvider exhausted during retry.");
}
item = this.item;
context.setAttribute(ITEM, item);
}
return item;
@@ -157,9 +143,6 @@ public class ItemReaderRetryCallback implements RetryCallback {
if (keyGenerator != null) {
return keyGenerator;
}
if (reader instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) reader;
}
if (writer instanceof ItemKeyGenerator) {
return (ItemKeyGenerator) writer;
}
@@ -178,9 +161,6 @@ public class ItemReaderRetryCallback implements RetryCallback {
if (failedItemIdentifier != null) {
return failedItemIdentifier;
}
if (reader instanceof FailedItemIdentifier) {
return (FailedItemIdentifier) reader;
}
if (writer instanceof FailedItemIdentifier) {
return (FailedItemIdentifier) writer;
}
@@ -198,22 +178,10 @@ public class ItemReaderRetryCallback implements RetryCallback {
if (recoverer != null) {
return recoverer;
}
if (reader instanceof ItemRecoverer) {
return (ItemRecoverer) reader;
}
if (writer instanceof ItemRecoverer) {
return (ItemRecoverer) writer;
}
return null;
}
/**
* Public getter for the {@link ItemReader}.
*
* @return the {@link ItemReader} instance.
*/
public ItemReader getReader() {
return reader;
}
}

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

@@ -26,25 +26,25 @@ import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.RetryPolicy;
import org.springframework.batch.retry.TerminatedRetryException;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.context.RetryContextSupport;
import org.springframework.util.Assert;
/**
* A {@link RetryPolicy} that detects an {@link ItemReaderRetryCallback} when it
* A {@link RetryPolicy} that detects an {@link ItemWriterRetryCallback} 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 ItemReaderRetryCallback} otherwise an exception will be thrown when
* of {@link ItemWriterRetryCallback} otherwise an exception will be thrown when
* the context is created.
*
* @author Dave Syer
*
*/
public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
public class ItemWriterRetryPolicy extends AbstractStatefulRetryPolicy {
protected Log logger = LogFactory.getLog(getClass());
public static final String EXHAUSTED = ItemReaderRetryPolicy.class.getName() + ".EXHAUSTED";
public static final String EXHAUSTED = ItemWriterRetryPolicy.class.getName() + ".EXHAUSTED";
private RetryPolicy delegate;
@@ -53,7 +53,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
*
* @param delegate
*/
public ItemReaderRetryPolicy(RetryPolicy delegate) {
public ItemWriterRetryPolicy(RetryPolicy delegate) {
super();
this.delegate = delegate;
}
@@ -62,7 +62,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
* Default constructor. Creates a new {@link SimpleRetryPolicy} for the
* delegate.
*/
public ItemReaderRetryPolicy() {
public ItemWriterRetryPolicy() {
this(new SimpleRetryPolicy());
}
@@ -96,7 +96,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
/**
* Create a new context for the execution of the callback, which must be an
* instance of {@link ItemReaderRetryCallback}.
* instance of {@link ItemWriterRetryCallback}.
*
* @see org.springframework.batch.retry.RetryPolicy#open(org.springframework.batch.retry.RetryCallback,
* RetryContext)
@@ -105,8 +105,8 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
* type.
*/
public RetryContext open(RetryCallback callback, RetryContext parent) {
Assert.state(callback instanceof ItemReaderRetryCallback, "Callback must be ItemProviderRetryCallback");
ItemReaderRetryContext context = new ItemReaderRetryContext((ItemReaderRetryCallback) callback, parent);
Assert.state(callback instanceof ItemWriterRetryCallback, "Callback must be ItemProviderRetryCallback");
ItemWriterRetryContext context = new ItemWriterRetryContext((ItemWriterRetryCallback) callback, parent);
context.open(callback, null);
return context;
}
@@ -132,7 +132,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
return ((RetryPolicy) context).handleRetryExhausted(context);
}
private class ItemReaderRetryContext extends RetryContextSupport implements RetryPolicy {
private class ItemWriterRetryContext extends RetryContextSupport implements RetryPolicy {
private Object item;
@@ -145,7 +145,7 @@ public class ItemReaderRetryPolicy extends AbstractStatefulRetryPolicy {
private FailedItemIdentifier failedItemIdentifier;
public ItemReaderRetryContext(ItemReaderRetryCallback callback, RetryContext parent) {
public ItemWriterRetryContext(ItemWriterRetryCallback callback, RetryContext parent) {
super(parent);
item = callback.next(this);
this.recoverer = callback.getRecoverer();

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

@@ -31,7 +31,7 @@ import org.springframework.batch.retry.context.RetryContextSupport;
import org.springframework.batch.retry.policy.NeverRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
public class ItemReaderRetryCallbackTests extends TestCase {
public class ItemWriterRetryCallbackTests extends TestCase {
List calls = new ArrayList();
@@ -41,12 +41,14 @@ public class ItemReaderRetryCallbackTests extends TestCase {
ListItemReaderRecoverer provider;
ItemReaderRetryCallback callback;
ItemWriterRetryCallback callback;
private AbstractItemWriter writer;
protected void setUp() throws Exception {
super.setUp();
template = new RetryTemplate();
provider = new ListItemReaderRecoverer(Arrays.asList(new String[] { "foo", "bar" })) {
provider = new ListItemReaderRecoverer(Arrays.asList(new String[] { "foo" })) {
public boolean recover(Object data, Throwable cause) {
count++;
calls.add(data);
@@ -56,14 +58,15 @@ public class ItemReaderRetryCallbackTests extends TestCase {
return "key" + (count++);
}
};
callback = new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
writer = new AbstractItemWriter() {
public void write(Object data) {
count++;
if (data.equals("bar")) {
throw new IllegalStateException("Bar detected");
}
}
});
};
callback = new ItemWriterRetryCallback("foo", writer);
}
public void testDoWithRetrySuccessfulFirstTime() throws Exception {
@@ -71,30 +74,15 @@ public class ItemReaderRetryCallbackTests extends TestCase {
assertEquals(1, count);
}
public void testDataExhausted() throws Exception {
provider.read();
provider.read(); // line up a null data item...
try {
template.execute(callback);
}
catch (RetryException e) {
fail("Unexpected RetryException");
}
// The item is null, and is not processed:
assertEquals(0, count);
}
public void testContextInitializedWithItemAndCanRetry() throws Exception {
// We can use the policy to intercept the context and do something with
// the item...
provider.read(); // line up an unsuccessful call...
callback = new ItemWriterRetryCallback("bar", writer);
assertEquals(0, calls.size());
template.setRetryPolicy(new NeverRetryPolicy() {
public boolean canRetry(RetryContext context) {
// ...register the failed item
calls.add("item(" + count + ")=" + context.getAttribute(ItemReaderRetryCallback.ITEM));
calls.add("item(" + count + ")=" + context.getAttribute(ItemWriterRetryCallback.ITEM));
// Do not call the base class method - the attempt counts as
// successful now
if (count < 2) // only retry once
@@ -120,12 +108,12 @@ public class ItemReaderRetryCallbackTests extends TestCase {
public void testContextInitializedWithItemAndRegisterThrowable() throws Exception {
// We can use the policy to intercept the context and do something with
// the item...
provider.read(); // line up an unsuccessful call...
callback = new ItemWriterRetryCallback("bar", writer);
assertEquals(0, calls.size());
template.setRetryPolicy(new NeverRetryPolicy() {
public void registerThrowable(RetryContext context, Throwable throwable) throws TerminatedRetryException {
// ...register the failed item
calls.add("item=" + context.getAttribute(ItemReaderRetryCallback.ITEM));
calls.add("item=" + context.getAttribute(ItemWriterRetryCallback.ITEM));
// Call the base class method so that the next attempt is a
// failure.
super.registerThrowable(context, throwable);
@@ -156,11 +144,12 @@ public class ItemReaderRetryCallbackTests extends TestCase {
}
public void testGetKey() throws Exception {
callback.setKeyGenerator(provider);
assertEquals("key0", callback.getKeyGenerator().getKey("foo"));
}
public void testRecoverWithoutSession() throws Exception {
callback.getRecoverer().recover("foo", null);
provider.recover("foo", null);
assertEquals(1, count);
assertEquals(1, calls.size());
}

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

@@ -33,13 +33,13 @@ import org.springframework.batch.repeat.support.RepeatSynchronizationManager;
import org.springframework.batch.retry.ListItemReaderRecoverer;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.context.RetryContextSupport;
import org.springframework.batch.retry.support.RetryTemplate;
public class ItemReaderRetryPolicyTests extends TestCase {
public class ItemWriterRetryPolicyTests extends TestCase {
private ItemReaderRetryPolicy policy = new ItemReaderRetryPolicy();
private ItemWriterRetryPolicy policy = new ItemWriterRetryPolicy();
private ListItemReaderRecoverer reader;
@@ -67,7 +67,7 @@ public class ItemReaderRetryPolicyTests extends TestCase {
}
public void testOpenSunnyDay() throws Exception {
RetryContext context = policy.open(new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -76,9 +76,6 @@ public class ItemReaderRetryPolicyTests extends TestCase {
assertNotNull(context);
// we haven't called the processor yet...
assertEquals(0, count);
// but the provider has been accessed:
assertEquals("foo", reader.read());
assertEquals("bar", reader.read());
}
public void testOpenWithWrongCallbackType() {
@@ -98,7 +95,7 @@ public class ItemReaderRetryPolicyTests extends TestCase {
public void testCanRetry() {
policy.setDelegate(new AlwaysRetryPolicy());
RetryContext context = policy.open(new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
}
@@ -111,7 +108,7 @@ public class ItemReaderRetryPolicyTests extends TestCase {
public void testRegisterThrowable() {
policy.setDelegate(new NeverRetryPolicy());
RetryContext context = policy.open(new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -124,7 +121,7 @@ public class ItemReaderRetryPolicyTests extends TestCase {
public void testClose() throws Exception {
policy.setDelegate(new NeverRetryPolicy());
RetryContext context = policy.open(new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -137,13 +134,10 @@ public class ItemReaderRetryPolicyTests extends TestCase {
// still can't retry, even if policy is closed
// (not that this would happen in practice)...
assertFalse(policy.canRetry(context));
// The provider has been accessed only once:
assertEquals("foo", reader.read());
assertEquals("bar", reader.read());
}
public void testOpenTwice() throws Exception {
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -165,19 +159,16 @@ public class ItemReaderRetryPolicyTests extends TestCase {
assertFalse(policy.canRetry(context));
policy.close(context);
// The provider has been accessed twice, so this
// mimics a message receive by repeating the value of the first
// message...
assertEquals("bar", reader.read());
}
public void testRecover() throws Exception {
policy = new ItemReaderRetryPolicy();
policy = new ItemWriterRetryPolicy();
policy.setDelegate(new SimpleRetryPolicy(1));
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
}
});
callback.setRecoverer(reader);
RetryContext context = policy.open(callback, null);
assertNotNull(context);
assertTrue(policy.canRetry(context));
@@ -202,22 +193,24 @@ public class ItemReaderRetryPolicyTests extends TestCase {
}
public void testFailedItemIdentifier() throws Exception {
policy = new ItemReaderRetryPolicy();
policy = new ItemWriterRetryPolicy();
policy.setDelegate(new SimpleRetryPolicy(1));
MockFailedItemProvider provider = new MockFailedItemProvider(Collections.EMPTY_LIST);
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(provider, null);
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", null);
callback.setFailedItemIdentifier(provider);
policy.open(callback, null);
assertEquals(1, provider.hasFailedCount);
}
public void testRecoverWithTemplate() throws Exception {
policy = new ItemReaderRetryPolicy();
policy = new ItemWriterRetryPolicy();
policy.setDelegate(new SimpleRetryPolicy(1));
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
throw new RuntimeException("Barf!");
}
});
callback.setRecoverer(reader);
RetryTemplate template = new RetryTemplate();
template.setRetryPolicy(policy);
Object result = null;
@@ -236,7 +229,7 @@ public class ItemReaderRetryPolicyTests extends TestCase {
}
public void testExhaustedClearsHistoryAfterLastAttempt() throws Exception {
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -262,9 +255,9 @@ public class ItemReaderRetryPolicyTests extends TestCase {
}
public void testRetryCount() throws Exception {
policy = new ItemReaderRetryPolicy();
policy = new ItemWriterRetryPolicy();
policy.setDelegate(new SimpleRetryPolicy(1));
RetryContext context = policy.open(new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
RetryContext context = policy.open(new ItemWriterRetryCallback("foo", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
@@ -279,14 +272,14 @@ public class ItemReaderRetryPolicyTests extends TestCase {
}
public void testRetryCountPreservedBetweenRetries() throws Exception {
ItemReaderRetryCallback callback = new ItemReaderRetryCallback(reader, new AbstractItemWriter() {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback("bar", new AbstractItemWriter() {
public void write(Object data) {
count++;
list.add(data);
}
});
policy = new ItemReaderRetryPolicy();
policy = new ItemWriterRetryPolicy();
policy.setDelegate(new SimpleRetryPolicy(1));
RetryContext context = policy.open(callback, null);
assertNotNull(context);

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

@@ -23,15 +23,14 @@ import javax.sql.DataSource;
import org.springframework.batch.item.AbstractItemReader;
import org.springframework.batch.item.AbstractItemWriter;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.policy.ItemWriterRetryPolicy;
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -49,7 +48,7 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
private RepeatTemplate repeatTemplate;
private ItemReader provider;
private ItemReaderRecoverer provider;
private JdbcTemplate jdbcTemplate;
@@ -113,18 +112,7 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
public void testExternalRetryRecoveryInBatch() throws Exception {
assertInitialState();
retryTemplate.setRetryPolicy(new ItemReaderRetryPolicy(new SimpleRetryPolicy(1)));
final ItemReaderRetryCallback callback = new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
public void write(final Object text) {
// No need for transaction here: the whole batch will roll
// back. When it comes back for recovery this code is not
// executed...
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
throw new RuntimeException("Rollback!");
}
});
retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy(new SimpleRetryPolicy(1)));
repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2));
@@ -139,7 +127,30 @@ public class ExternalRetryInBatchTests extends AbstractDependencyInjectionSpring
repeatTemplate.iterate(new RepeatCallback() {
public ExitStatus doInIteration(RepeatContext context) throws Exception {
return new ExitStatus(retryTemplate.execute(callback) != null);
Object item = provider.read();
if (item==null) {
return ExitStatus.FINISHED;
}
ItemWriterRetryCallback callback = new ItemWriterRetryCallback(item, new AbstractItemWriter() {
public void write(final Object text) {
// No need for transaction here: the whole batch will roll
// back. When it comes back for recovery this code is not
// executed...
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
throw new RuntimeException("Rollback!");
}
});
callback.setRecoverer(provider);
retryTemplate.execute(callback);
return ExitStatus.CONTINUABLE;
}
});

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

@@ -23,10 +23,9 @@ import javax.sql.DataSource;
import org.springframework.batch.item.AbstractItemReader;
import org.springframework.batch.item.AbstractItemWriter;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.policy.ItemWriterRetryPolicy;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
@@ -42,7 +41,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
private RetryTemplate retryTemplate;
private ItemReader provider;
private ItemReaderRecoverer provider;
private JdbcTemplate jdbcTemplate;
@@ -102,9 +101,9 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
assertInitialState();
retryTemplate.setRetryPolicy(new ItemReaderRetryPolicy());
retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy());
final ItemReaderRetryCallback callback = new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
final AbstractItemWriter writer = new AbstractItemWriter() {
public void write(final Object text) {
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
@@ -113,12 +112,13 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
}
}
});
};
try {
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
try {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), writer);
return retryTemplate.execute(callback);
} catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
@@ -138,6 +138,7 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
new TransactionTemplate(transactionManager).execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
try {
ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), writer);
return retryTemplate.execute(callback);
} catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
@@ -164,15 +165,16 @@ public class ExternalRetryTests extends AbstractDependencyInjectionSpringContext
assertInitialState();
retryTemplate.setRetryPolicy(new ItemReaderRetryPolicy());
retryTemplate.setRetryPolicy(new ItemWriterRetryPolicy());
final ItemReaderRetryCallback callback = new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
final ItemWriterRetryCallback callback = new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() {
public void write(final Object text) {
jdbcTemplate.update("INSERT into T_FOOS (id,name,foo_date) values (?,?,null)", new Object[] {
new Integer(list.size()), text });
throw new RuntimeException("Rollback!");
}
});
callback.setRecoverer(provider);
Object result = "start";

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

@@ -23,7 +23,7 @@ import org.springframework.batch.item.AbstractItemWriter;
import org.springframework.batch.item.jms.JmsItemReader;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
@@ -144,7 +144,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
provider.setJmsTemplate(jmsTemplate);
jmsTemplate.setDefaultDestinationName("queue");
retryTemplate.execute(new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
retryTemplate.execute(new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() {
public void write(final Object text) {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);

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

@@ -23,7 +23,7 @@ import org.springframework.batch.item.AbstractItemWriter;
import org.springframework.batch.item.jms.JmsItemReader;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
import org.springframework.batch.retry.callback.ItemWriterRetryCallback;
import org.springframework.batch.retry.support.RetryTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.AbstractTransactionalDataSourceSpringContextTests;
@@ -140,7 +140,7 @@ public class SynchronousTests extends AbstractTransactionalDataSourceSpringConte
provider.setJmsTemplate(jmsTemplate);
jmsTemplate.setDefaultDestinationName("queue");
retryTemplate.execute(new ItemReaderRetryCallback(provider, new AbstractItemWriter() {
retryTemplate.execute(new ItemWriterRetryCallback(provider.read(), new AbstractItemWriter() {
public void write(final Object text) {
TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);