From 5b056e812667a49e35f04775279f8e1f43f66f4e Mon Sep 17 00:00:00 2001
From: Michael Minella <mminella@pivotal.io>
Date: Tue, 23 Sep 2014 14:12:48 -0500
Subject: [PATCH] Handle scenario where scanned item is filtered out

BATCH-2302 documents a scenario where an item throws an exception in a
fault tollerant step in the write, then in the process, throws an
exception as well.  This leads to an infinite loop.  To address this,
prior to attempting to write items during scanning we need to validate
that the there are items to be written (we were not which was causing a
NoSuchElementException when we did inputs.next().

This commit also removes an eronous System.out left in the
CoreNamespaceUtils.
---
 .../configuration/xml/CoreNamespaceUtils.java |   1 -
 .../item/FaultTolerantChunkProcessor.java     |   4 +-
 .../step/skip/ReprocessExceptionTests.java    | 123 ++++++++++++++++++
 ...syncChunkOrientedStepIntegrationTests.java |   2 -
 .../ChunkOrientedStepIntegrationTests.java    |   2 -
 .../src/test/resources/data/person.csv        |   5 +
 .../skip/ReprocessExceptionTests-context.xml  |  47 +++++++
 7 files changed, 176 insertions(+), 8 deletions(-)
 create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/ReprocessExceptionTests.java
 create mode 100644 spring-batch-core/src/test/resources/data/person.csv
 create mode 100644 spring-batch-core/src/test/resources/org/springframework/batch/core/step/skip/ReprocessExceptionTests-context.xml

diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/CoreNamespaceUtils.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/CoreNamespaceUtils.java
index 7492990e7..b740d6312 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/CoreNamespaceUtils.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/CoreNamespaceUtils.java
@@ -58,7 +58,6 @@ public class CoreNamespaceUtils {
 	private static final String CORE_NAMESPACE_POST_PROCESSOR_CLASS_NAME = "org.springframework.batch.core.configuration.xml.CoreNamespacePostProcessor";
 
 	public static void autoregisterBeansForNamespace(ParserContext parserContext, Object source) {
-		System.out.println("******** CoreNamespaceUtils is called");
 		checkForStepScope(parserContext, source);
 		checkForJobScope(parserContext, source);
 		addRangePropertyEditor(parserContext);
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
index 2458f1a1f..4088277e7 100755
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
@@ -320,7 +320,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 		RetryCallback<Object, Exception> retryCallback = new RetryCallback<Object, Exception>() {
 			@Override
 			public Object doWithRetry(RetryContext context) throws Exception {
-
 				contextHolder.set(context);
 
 				if (!data.scanning()) {
@@ -395,7 +394,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 
 				@Override
 				public Object recover(RetryContext context) throws Exception {
-
 					/*
 					 * If the last exception was not skippable we don't need to
 					 * do any scanning. We can just bomb out with a retry
@@ -564,7 +562,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 				logger.debug("Scanning for failed item on write: " + inputs);
 			}
 		}
-		if (outputs.isEmpty()) {
+		if (outputs.isEmpty() || inputs.isEmpty()) {
 			data.scanning(false);
 			inputs.setBusy(false);
 			chunkMonitor.resetOffset();
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/ReprocessExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/ReprocessExceptionTests.java
new file mode 100644
index 000000000..fab91ca4c
--- /dev/null
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/skip/ReprocessExceptionTests.java
@@ -0,0 +1,123 @@
+/*
+ * Copyright 2014 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.core.step.skip;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * @author mminella
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class ReprocessExceptionTests {
+
+	@Autowired
+	public Job job;
+
+	@Autowired
+	public JobLauncher jobLauncher;
+
+	@Test
+	public void testReprocessException() throws Exception {
+		JobExecution execution = jobLauncher.run(job, new JobParametersBuilder().toJobParameters());
+
+		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
+	}
+
+	public static class PersonProcessor implements ItemProcessor<Person, Person> {
+
+		private String mostRecentFirstName;
+
+		@Override
+		public Person process(final Person person) throws Exception {
+			if (person.getFirstName().equals(mostRecentFirstName)) {
+				throw new RuntimeException("throwing a exception during process after a rollback");
+			}
+			mostRecentFirstName = person.getFirstName();
+
+			final String firstName = person.getFirstName().toUpperCase();
+			final String lastName = person.getLastName().toUpperCase();
+
+			final Person transformedPerson = new Person(firstName, lastName);
+
+			System.out.println("Converting (" + person + ") into (" + transformedPerson + ")");
+
+			return transformedPerson;
+		}
+	}
+
+	public static class PersonItemWriter implements ItemWriter<Person> {
+		@Override
+		public void write(List<? extends Person> persons) throws Exception {
+			for (Person person : persons) {
+				System.out.println(person.getFirstName() + " " + person.getLastName());
+				if (person.getFirstName().equals("JANE")) {
+					throw new RuntimeException("jane doe write exception causing rollback");
+				}
+			}
+		}
+	}
+
+	public static class Person {
+		private String lastName;
+		private String firstName;
+
+		public Person() {
+
+		}
+
+		public Person(String firstName, String lastName) {
+			this.firstName = firstName;
+			this.lastName = lastName;
+		}
+
+		public void setFirstName(String firstName) {
+			this.firstName = firstName;
+		}
+
+		public String getFirstName() {
+			return firstName;
+		}
+
+		public String getLastName() {
+			return lastName;
+		}
+
+		public void setLastName(String lastName) {
+			this.lastName = lastName;
+		}
+
+		@Override
+		public String toString() {
+			return "firstName: " + firstName + ", lastName: " + lastName;
+		}
+	}
+}
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java
index 47d20da2d..18212ff73 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java
@@ -18,7 +18,6 @@ package org.springframework.batch.core.step.tasklet;
 import org.apache.commons.dbcp.BasicDataSource;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.batch.core.BatchStatus;
@@ -120,7 +119,6 @@ public class AsyncChunkOrientedStepIntegrationTests {
 	}
 
 	@Test
-	@Ignore //FIXME
 	public void testStatus() throws Exception {
 
 		step.setTasklet(new TestingChunkOrientedTasklet<String>(getReader(new String[] { "a", "b", "c", "a", "b", "c",
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java
index 260f02f70..680136d90 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java
@@ -16,7 +16,6 @@
 package org.springframework.batch.core.step.tasklet;
 
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.batch.core.BatchStatus;
@@ -90,7 +89,6 @@ public class ChunkOrientedStepIntegrationTests {
 
 	@SuppressWarnings("serial")
 	@Test
-	@Ignore //FIXME
 	public void testStatusForCommitFailedException() throws Exception {
 
 		step.setTasklet(new TestingChunkOrientedTasklet<String>(getReader(new String[] { "a", "b", "c" }),
diff --git a/spring-batch-core/src/test/resources/data/person.csv b/spring-batch-core/src/test/resources/data/person.csv
new file mode 100644
index 000000000..cead90f5a
--- /dev/null
+++ b/spring-batch-core/src/test/resources/data/person.csv
@@ -0,0 +1,5 @@
+Jill,Doe
+Joe,Doe
+Justin,Doe
+Jane,Doe
+John,Doe
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/step/skip/ReprocessExceptionTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/step/skip/ReprocessExceptionTests-context.xml
new file mode 100644
index 000000000..0566b34b2
--- /dev/null
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/step/skip/ReprocessExceptionTests-context.xml
@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   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.xsd
+	   			http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd">
+
+	<bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
+		<property name="jobRepository" ref="jobRepository"/>
+	</bean>
+
+	<bean id="reader" class="org.springframework.batch.item.file.FlatFileItemReader">
+		<property name="resource" value="classpath:/data/person.csv"/>
+		<property name="lineMapper">
+			<bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
+				<property name="lineTokenizer">
+					<bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
+						<property name="names" value="firstName,lastName"/>
+					</bean>
+				</property>
+				<property name="fieldSetMapper">
+					<bean class="org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper">
+						<property name="targetType" value="org.springframework.batch.core.step.skip.ReprocessExceptionTests$Person"/>
+					</bean>
+				</property>
+			</bean>
+		</property>
+	</bean>
+
+	<bean id="processor" class="org.springframework.batch.core.step.skip.ReprocessExceptionTests$PersonProcessor"/>
+
+	<bean id="writer" class="org.springframework.batch.core.step.skip.ReprocessExceptionTests$PersonItemWriter"/>
+
+	<job id="job" xmlns="http://www.springframework.org/schema/batch">
+		<step id="step1">
+			<tasklet>
+				<chunk reader="reader" processor="processor" writer="writer" commit-interval="1">
+					<skip-policy>
+						<bean class="org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy" xmlns="http://www.springframework.org/schema/beans"/>
+					</skip-policy>
+				</chunk>
+			</tasklet>
+		</step>
+	</job>
+
+	<bean id="jobRepository" class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean"/>
+	<bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager"/>
+</beans>
\ No newline at end of file