RESOLVED - issue BATCH-268: retry configuration for ItemOrientedTasklet

http://jira.springframework.org/browse/BATCH-268

The retry only works if you have an ItemStream as an ItemReader (so that you get a proper rollback).  Applied the patch and fixed it up to fit the new m5 way of looking at things.

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/AbstractStep.java

@@ -27,6 +27,7 @@ import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.stream.SimpleStreamManager;
 import org.springframework.batch.item.stream.StreamManager;
 import org.springframework.batch.repeat.exception.handler.ExceptionHandler;
+import org.springframework.batch.retry.RetryPolicy;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.Assert;
@@ -42,6 +43,8 @@ public abstract class AbstractStep extends StepSupport implements InitializingBe
 	private int skipLimit = 0;
 
 	private ExceptionHandler exceptionHandler;
+	
+	private RetryPolicy retryPolicy;
 
 	private JobRepository jobRepository;
 
@@ -71,6 +74,22 @@ public abstract class AbstractStep extends StepSupport implements InitializingBe
 		super(name);
 	}
 
+	/**
+	 * Public getter for the {@link RetryPolicy}.
+	 * @return the {@link RetryPolicy}
+	 */
+	public RetryPolicy getRetryPolicy() {
+		return retryPolicy;
+	}
+
+	/**
+	 * Public setter for the {@link RetryPolicy}.
+	 * @param retryPolicy the {@link RetryPolicy} to set
+	 */
+	public void setRetryPolicy(RetryPolicy retryPolicy) {
+		this.retryPolicy = retryPolicy;
+	}
+
 	public ExceptionHandler getExceptionHandler() {
 		return exceptionHandler;
 	}
@@ -177,6 +196,7 @@ public abstract class AbstractStep extends StepSupport implements InitializingBe
 		executor.setItemWriter(itemWriter);
 		executor.setItemRecoverer(itemRecoverer);
 		executor.setRepository(jobRepository);
+		executor.setRetryPolicy(retryPolicy);
 		executor.setStreamManager(manager);
 		try {
 			executor.afterPropertiesSet();

spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java

@@ -315,9 +315,6 @@ public class SimpleStepExecutor implements InitializingBean {
 
 					try {
 
-						/*
-						 * New transaction obtained, resynchronize TransactionSynchronization objects
-						 */
 						result = processChunk(step, contribution);
 
 						// TODO: check that stepExecution can

spring-batch-infrastructure/src/main/java/org/springframework/batch/io/exception/ReadFailureException.java

@@ -16,6 +16,8 @@
 
 package org.springframework.batch.io.exception;
 
+import org.springframework.batch.item.ItemReader;
+
 /**
  * This exception indicates an error encountered while reading.  It should generally
  * be thrown by classes that implement the {@link ItemReader} interface.

spring-batch-infrastructure/src/main/java/org/springframework/batch/io/exception/WriteFailureException.java

@@ -16,6 +16,8 @@
 
 package org.springframework.batch.io.exception;
 
+import org.springframework.batch.item.ItemWriter;
+
 /**
  * Exception thrown after encountering an error during a write.  It should
  * generally be thrown by classes that implement the {@link ItemWriter} 

spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java

@@ -23,8 +23,8 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.io.Skippable;
 import org.springframework.batch.io.file.separator.LineReader;
-import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ExecutionAttributes;
+import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.exception.StreamException;
 
 /**

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/AbstractItemReaderRecoverer.java

@@ -13,9 +13,11 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.springframework.batch.item;
+package org.springframework.batch.item.reader;
+
+import org.springframework.batch.item.ItemRecoverer;
+import org.springframework.batch.item.KeyedItemReader;
 
-import org.springframework.batch.item.reader.AbstractItemReader;
 
 /**
  * @author Dave Syer

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/reader/DelegatingItemReader.java

@@ -17,9 +17,9 @@
 package org.springframework.batch.item.reader;
 
 import org.springframework.batch.io.Skippable;
+import org.springframework.batch.item.ExecutionAttributes;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
-import org.springframework.batch.item.ExecutionAttributes;
 import org.springframework.batch.item.exception.StreamException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/ItemStreamAdapter.java

@@ -15,8 +15,8 @@
  */
 package org.springframework.batch.item.stream;
 
-import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ExecutionAttributes;
+import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.exception.StreamException;
 
 /**

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/callback/NestedRepeatCallback.java

@@ -16,10 +16,10 @@
 
 package org.springframework.batch.repeat.callback;
 
+import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatCallback;
 import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.RepeatOperations;
-import org.springframework.batch.repeat.ExitStatus;
 
 /**
  * Callback that delegates to another callback, via a {@link RepeatOperations} instance.

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/CompletionPolicySupport.java

@@ -16,9 +16,9 @@
 
 package org.springframework.batch.repeat.policy;
 
+import org.springframework.batch.repeat.CompletionPolicy;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
-import org.springframework.batch.repeat.CompletionPolicy;
 import org.springframework.batch.repeat.context.RepeatContextSupport;
 
 /**

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/CompositeCompletionPolicy.java

@@ -19,9 +19,9 @@ package org.springframework.batch.repeat.policy;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.springframework.batch.repeat.CompletionPolicy;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatContext;
-import org.springframework.batch.repeat.CompletionPolicy;
 import org.springframework.batch.repeat.context.RepeatContextSupport;
 
 /**

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/policy/DefaultResultCompletionPolicy.java

@@ -16,9 +16,9 @@
 
 package org.springframework.batch.repeat.policy;
 
-import org.springframework.batch.repeat.RepeatContext;
 import org.springframework.batch.repeat.CompletionPolicy;
 import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.repeat.RepeatContext;
 
 /**
  * Very simple {@link CompletionPolicy} that bases its decision on the result of

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

@@ -21,8 +21,8 @@ import java.util.List;
 
 import javax.sql.DataSource;
 
-import org.springframework.batch.item.AbstractItemReaderRecoverer;
 import org.springframework.batch.item.KeyedItemReader;
+import org.springframework.batch.item.reader.AbstractItemReaderRecoverer;
 import org.springframework.batch.item.writer.AbstractItemWriter;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.repeat.RepeatCallback;

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

@@ -21,8 +21,8 @@ import java.util.List;
 
 import javax.sql.DataSource;
 
-import org.springframework.batch.item.AbstractItemReaderRecoverer;
 import org.springframework.batch.item.KeyedItemReader;
+import org.springframework.batch.item.reader.AbstractItemReaderRecoverer;
 import org.springframework.batch.item.writer.AbstractItemWriter;
 import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
 import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;

spring-batch-samples/.springBeans

@@ -31,6 +31,7 @@
 		<config>src/main/resources/jobs/delegatingJob.xml</config>
 		<config>src/main/resources/jobs/parallelJob.xml</config>
 		<config>src/main/resources/jobs/rollbackJob.xml</config>
+		<config>src/main/resources/jobs/retrySample.xml</config>
 	</configs>
 	<configSets>
 		<configSet>
@@ -235,5 +236,16 @@
 				<config>src/main/resources/jobs/tradeJobIo.xml</config>
 			</configs>
 		</configSet>
+		<configSet>
+			<name><![CDATA[retry]]></name>
+			<allowBeanDefinitionOverriding>true</allowBeanDefinitionOverriding>
+			<incomplete>false</incomplete>
+			<configs>
+				<config>src/main/resources/jobs/retrySample.xml</config>
+				<config>src/main/resources/simple-container-definition.xml</config>
+				<config>src/main/resources/data-source-context.xml</config>
+				<config>src/main/resources/data-source-context-init.xml</config>
+			</configs>
+		</configSet>
 	</configSets>
 </beansProjectDescription>

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/GeneratingItemReader.java

@@ -0,0 +1,109 @@
+package org.springframework.batch.sample.item.reader;
+
+import java.math.BigDecimal;
+
+import org.springframework.batch.item.ExecutionAttributes;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.exception.MarkFailedException;
+import org.springframework.batch.item.exception.ResetFailedException;
+import org.springframework.batch.item.exception.StreamException;
+import org.springframework.batch.item.reader.AbstractItemReaderRecoverer;
+import org.springframework.batch.sample.domain.Trade;
+
+/**
+ * Generates configurable number of {@link Trade} items.
+ * 
+ * @author Robert Kasanicky
+ */
+public class GeneratingItemReader extends AbstractItemReaderRecoverer implements ItemStream {
+
+	private int limit = 1;
+	
+	private int counter = 0;
+
+	private int marked;
+	
+	public Object read() throws Exception {
+		if (counter < limit) {
+			counter++;
+			return new Trade(
+					"isin" + counter, 
+					counter, 
+					new BigDecimal(counter), 
+					"customer" + counter);
+		}
+		return null;
+	}
+
+	/**
+	 * @param limit number of items that will be generated
+	 * (null returned on consecutive calls).
+	 */
+	public void setLimit(int limit) {
+		this.limit = limit;
+	}
+
+	public int getCounter() {
+		return counter;
+	}
+
+	public int getLimit() {
+		return limit;
+	}
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemRecoverer#recover(java.lang.Object, java.lang.Throwable)
+	 */
+	public boolean recover(Object data, Throwable cause) {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#close()
+	 */
+	public void close() throws StreamException {
+		// TODO Auto-generated method stub
+		
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#isMarkSupported()
+	 */
+	public boolean isMarkSupported() {
+		return true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#mark()
+	 */
+	public void mark() throws MarkFailedException {
+		this.marked = this.counter;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#open()
+	 */
+	public void open() throws StreamException {
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#reset()
+	 */
+	public void reset() throws ResetFailedException {
+		this.counter = this.marked;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemStream#restoreFrom(org.springframework.batch.item.ExecutionAttributes)
+	 */
+	public void restoreFrom(ExecutionAttributes context) {
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.ExecutionAttributesProvider#getExecutionAttributes()
+	 */
+	public ExecutionAttributes getExecutionAttributes() {
+		return new ExecutionAttributes();
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/writer/RetrySampleItemWriter.java

@@ -0,0 +1,29 @@
+package org.springframework.batch.sample.item.writer;
+
+import org.springframework.batch.item.ItemWriter;
+
+/**
+ * Simulates temporary output trouble - requires to
+ * retry 3 times to pass successfully.
+ * 
+ * @author Robert Kasanicky
+ */
+public class RetrySampleItemWriter implements ItemWriter {
+
+	private int counter = 0;
+	
+	public void write(Object data) throws Exception {
+		counter++;
+		if (counter == 2 || counter == 3) {
+			throw new RuntimeException("Temporary error");
+		}
+	}
+
+	/**
+	 * @return number of times {@link #process(Object)} method was called.
+	 */
+	public int getCounter() {
+		return counter;
+	}
+
+}

spring-batch-samples/src/main/resources/jobs/retrySample.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:aop="http://www.springframework.org/schema/aop"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xmlns:p="http://www.springframework.org/schema/p"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
+		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd">
+
+	<bean parent="stepScope" />
+	<bean parent="jobConfigurationRegistryBeanPostProcessor" />
+
+	<bean id="retrySample" parent="simpleJob">
+		<property name="steps">
+			<bean id="step1" parent="simpleStep">
+				<property name="itemReader" ref="itemGenerator" />
+				<property name="itemWriter" ref="itemWriter" />
+				<property name="retryPolicy">
+					<bean
+						class="org.springframework.batch.retry.policy.SimpleRetryPolicy">
+						<property name="maxAttempts" value="3" />
+						<property name="retryableExceptionClasses"
+							value="java.lang.Exception" />
+					</bean>
+				</property>
+				<property name="exceptionHandler">
+					<bean
+						class="org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHandler">
+						<property name="limit" value="5" />
+						<property name="type"
+							value="java.lang.Exception" />
+					</bean>
+				</property>
+			</bean>
+		</property>
+	</bean>
+
+	<bean id="testGenerator"
+		class="org.springframework.batch.sample.item.reader.GeneratingItemReader">
+		<property name="limit" value="10" />
+	</bean>
+
+	<bean id="itemGenerator" parent="testGenerator" scope="step" autowire-candidate="false">
+		<aop:scoped-proxy/>
+	</bean>
+
+	<bean id="itemWriter"
+		class="org.springframework.batch.sample.item.writer.RetrySampleItemWriter" />
+</beans>

spring-batch-samples/src/test/java/org/springframework/batch/sample/RetrySampleFunctionalTests.java

@@ -0,0 +1,34 @@
+package org.springframework.batch.sample;
+
+import org.springframework.batch.sample.item.reader.GeneratingItemReader;
+import org.springframework.batch.sample.item.writer.RetrySampleItemWriter;
+
+/**
+ * Checks that expected number of items have been processed.
+ * 
+ * @author Robert Kasanicky
+ */
+public class RetrySampleFunctionalTests extends AbstractValidatingBatchLauncherTests {
+
+	private GeneratingItemReader itemGenerator;
+	
+	private RetrySampleItemWriter itemProcessor;
+	
+	protected String[] getConfigLocations() {
+		return new String[] {"jobs/retrySample.xml"};
+	}
+	
+	protected void validatePostConditions() throws Exception {
+		//items processed = items read + 2 exceptions
+		assertEquals(itemGenerator.getLimit()+2, itemProcessor.getCounter());
+	}
+	
+	public void setItemGenerator(GeneratingItemReader itemGenerator) {
+		this.itemGenerator = itemGenerator;
+	}
+
+	public void setItemProcessor(RetrySampleItemWriter itemProcessor) {
+		this.itemProcessor = itemProcessor;
+	}
+
+}

spring-batch-samples/src/test/java/org/springframework/batch/sample/item/reader/GeneratingItemReaderTests.java

@@ -0,0 +1,31 @@
+package org.springframework.batch.sample.item.reader;
+
+import junit.framework.TestCase;
+
+/**
+ * Tests for {@link GeneratingItemReader}.
+ * 
+ * @author Robert Kasanicky
+ */
+public class GeneratingItemReaderTests extends TestCase {
+
+	private GeneratingItemReader reader = new GeneratingItemReader();
+	
+	/**
+	 * Generates a given number of not-null records,
+	 * consecutive calls return null.
+	 */
+	public void testRead() throws Exception {
+		int counter = 0;
+		int limit = 10;
+		reader.setLimit(limit);
+		
+		while (reader.read() != null) {
+			counter++;
+		}
+		
+		assertEquals(null, reader.read());
+		assertEquals(limit, counter);
+		assertEquals(counter, reader.getCounter());
+	}
+}

spring-batch-samples/src/test/java/org/springframework/batch/sample/item/writer/RetrySampleItemWriterTests.java

@@ -0,0 +1,35 @@
+package org.springframework.batch.sample.item.writer;
+
+import junit.framework.TestCase;
+
+/**
+ * Tests for {@link RetrySampleItemWriter}.
+ * 
+ * @author Robert Kasanicky
+ */
+public class RetrySampleItemWriterTests extends TestCase {
+
+	private RetrySampleItemWriter processor = new RetrySampleItemWriter();
+
+	/**
+	 * Processing throws exception on 2nd and 3rd call.
+	 */
+	public void testProcess() throws Exception {
+		Object item = null;
+		processor.write(item);
+
+		for (int i = 0; i < 2; i++) {
+			try {
+				processor.write(item);
+				fail();
+			}
+			catch (RuntimeException e) {
+				// expected
+			}
+		}
+		
+		processor.write(item);
+		
+		assertEquals(4, processor.getCounter());
+	}
+}