diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java index ce03916f5..556a5962f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBean.java @@ -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).
* - * @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); } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/validator/ValidatingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/validator/ValidatingItemReader.java index 1a8dbc9d6..48751a6ed 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/validator/ValidatingItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/validator/ValidatingItemReader.java @@ -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; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java similarity index 75% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java index ed9dd9836..e58cbcae2 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemReaderRetryCallback.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/callback/ItemWriterRetryCallback.java @@ -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; - } - } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java similarity index 90% rename from spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java rename to spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java index 8e1c1127d..03c564718 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicy.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicy.java @@ -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(); diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemReaderRetryCallbackTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java similarity index 85% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemReaderRetryCallbackTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java index 119c66bac..669421a22 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemReaderRetryCallbackTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/callback/ItemWriterRetryCallbackTests.java @@ -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()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java similarity index 82% rename from spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java rename to spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java index 7af6064ce..ace436d18 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemReaderRetryPolicyTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/retry/policy/ItemWriterRetryPolicyTests.java @@ -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); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java index 4cb8c018f..d6dc42009 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/jms/ExternalRetryInBatchTests.java @@ -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; + } }); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java index 2e662f4e9..ee4105f05 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/ExternalRetryTests.java @@ -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"; diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java index 9d0e7fb9b..24f213003 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/retry/jms/SynchronousTests.java @@ -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); diff --git a/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java b/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java index 2ae1bbf2b..5cb92b2c5 100644 --- a/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/retry/jms/SynchronousTests.java @@ -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);