OPEN - issue BATCH-1272: Write skips do not work in a multi-threaded step

Added some logging and a test case

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ExceptionThrowingItemHandlerStub.java

@@ -19,11 +19,16 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 /**
  * @author Dan Garrette
  * @since 2.0.1
  */
 public abstract class ExceptionThrowingItemHandlerStub<T> {
+	
+	protected Log logger = LogFactory.getLog(getClass());
 
 	private Collection<T> failures = Collections.emptyList();
 
@@ -44,10 +49,10 @@ public abstract class ExceptionThrowingItemHandlerStub<T> {
 	protected void checkFailure(T item) throws Exception {
 		if (isFailure(item)) {
 			if (runtimeException) {
-				throw new SkippableRuntimeException("Intended Failure");
+				throw new SkippableRuntimeException("Intended Failure: "+item);
 			}
 			else {
-				throw new SkippableException("Intended Failure");
+				throw new SkippableException("Intended Failure: "+item);
 			}
 		}
 	}

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRetryTests.java

@@ -314,13 +314,9 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,c,e,f"));
 		assertEquals(expectedOutput, written);
 
-		// [a, b, c, d, e, f, null]
-		assertEquals(7, provided.size());
-		// [a, b, b, b, b, b, c, d, d, d, d, d, e, f]
-		System.err.println(processed);
-		assertEquals(14, processed.size());
-		// [b, d]
-		assertEquals(2, recovered.size());
+		assertEquals("[a, b, c, d, e, f, null]", provided.toString());
+		assertEquals("[a, b, b, b, b, b, b, c, d, d, d, d, d, d, e, f]", processed.toString());
+		assertEquals("[b, d]", recovered.toString());
 	}
 
 	@Test
@@ -423,8 +419,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(1, provided.size());
 		// the failed items are tried up to the limit (but only precisely so if
 		// the commit interval is 1)
-		// [b, b, b, b]
-		assertEquals(4, processed.size());
+		assertEquals("[b, b, b, b, b]", processed.toString());
 		// []
 		assertEquals(0, recovered.size());
 		assertEquals(1, stepExecution.getReadCount());
@@ -474,9 +469,9 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 
 		assertEquals(0, stepExecution.getSkipCount());
 		// [b]
-		assertEquals(1, provided.size());
+		assertEquals("[b]", provided.toString());
 		// [b]
-		assertEquals(1, processed.size());
+		assertEquals("[b, b]", processed.toString());
 		// []
 		assertEquals(0, recovered.size());
 		assertEquals(1, stepExecution.getReadCount());
@@ -517,8 +512,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(0, stepExecution.getSkipCount());
 		// [b]
 		assertEquals(1, provided.size());
-		// [b, b, b, b]
-		assertEquals(4, processed.size());
+		assertEquals("[b, b, b, b, b]", processed.toString());
 		// []
 		assertEquals(0, recovered.size());
 		assertEquals(1, stepExecution.getReadCount());

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanRollbackTests.java

@@ -12,6 +12,7 @@ import java.util.List;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobExecution;
@@ -21,6 +22,7 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.transaction.interceptor.RollbackRuleAttribute;
 import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
 import org.springframework.transaction.interceptor.TransactionAttribute;
@@ -355,6 +357,24 @@ public class FaultTolerantStepFactoryBeanRollbackTests {
 				.toString());
 	}
 
+	@Test
+	@Ignore
+	public void testMultithreadedSkipInWriter() throws Exception {
+		writer.setFailures("1", "2", "3", "4", "5");
+		factory.setCommitInterval(3);
+		factory.setSkipLimit(10);
+		factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+
+		assertEquals("[]", writer.getCommitted().toString());
+		assertEquals("[]", processor.getCommitted().toString());
+		assertEquals(5, stepExecution.getSkipCount());
+	}
+
 	@Test
 	public void testMultipleSkipsInWriter() throws Exception {
 		writer.setFailures("2", "4");

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWriterStub.java

@@ -26,7 +26,7 @@ import org.springframework.batch.support.transaction.TransactionAwareProxyFactor
  * @since 2.0.1
  */
 public class SkipWriterStub<T> extends ExceptionThrowingItemHandlerStub<T> implements ItemWriter<T> {
-
+	
 	private List<T> written = new ArrayList<T>();
 
 	private List<T> committed = TransactionAwareProxyFactory.createTransactionalList();
@@ -46,6 +46,7 @@ public class SkipWriterStub<T> extends ExceptionThrowingItemHandlerStub<T> imple
 	}
 
 	public void write(List<? extends T> items) throws Exception {
+		logger.debug("Writing: "+items);
 		for (T item : items) {
 			written.add(item);
 			checkFailure(item);