RESOLVED: BATCH-1100. Change retry/skip logic to be more efficient (and limit to single-threaded readers)

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TaskletElementParserTests.java

@@ -87,17 +87,12 @@ public class TaskletElementParserTests {
 		Collection<ItemStream> streams = getStreams("s1", taskletElementParentAttributeParserTestsContext);
 		assertEquals(2, streams.size());
 		boolean c = false;
-		boolean d = false;
 		for (ItemStream o : streams) {
 			if (o instanceof CompositeItemStream) {
 				c = true;
 			}
-			else if (o instanceof TestReader) {
-				d = true;
-			}
 		}
 		assertTrue(c);
-		assertTrue(d);
 	}
 
 	@Test

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestReader.java

@@ -12,11 +12,11 @@ import org.springframework.batch.item.ParseException;
 import org.springframework.batch.item.UnexpectedInputException;
 
 public class TestReader extends AbstractTestComponent implements ItemReader<String>, ItemStream {
-	
+
 	private boolean opened = false;
 
 	List<String> items = null;
-	
+
 	{
 		List<String> l = new ArrayList<String>();
 		l.add("Item *** 1 ***");
@@ -31,28 +31,26 @@ public class TestReader extends AbstractTestComponent implements ItemReader<Stri
 	public void setOpened(boolean opened) {
 		this.opened = opened;
 	}
-	
-	public String read() throws Exception, UnexpectedInputException,
-			ParseException {
+
+	public String read() throws Exception, UnexpectedInputException, ParseException {
 		executed = true;
-		if (items.size() > 0) {
-			String item = items.remove(0); 
-			return item;
+		synchronized (items) {
+			if (items.size() > 0) {
+				String item = items.remove(0);
+				return item;
+			}
 		}
 		return null;
 	}
 
-	public void close()
-			throws ItemStreamException {
+	public void close() throws ItemStreamException {
 	}
 
-	public void open(ExecutionContext executionContext)
-			throws ItemStreamException {
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
 		opened = true;
 	}
 
-	public void update(ExecutionContext executionContext)
-			throws ItemStreamException {
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
 	}
 
 }

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

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2006-2007 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.item;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemStreamSupport;
+import org.springframework.batch.item.ParseException;
+import org.springframework.batch.item.UnexpectedInputException;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class ChunkMonitorTests {
+
+	/**
+	 * 
+	 */
+	private static final int CHUNK_SIZE = 5;
+
+	private ChunkMonitor monitor = new ChunkMonitor();
+
+	private int count = 0;
+
+	private boolean closed = false;
+
+	@Before
+	public void setUp() {
+		monitor.setItemReader(new ItemReader<String>() {
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				return "" + (count++);
+			}
+		});
+		monitor.setItemStream(new ItemStreamSupport() {
+			@Override
+			public void close() throws ItemStreamException {
+				closed = true;
+			}
+		});
+		monitor.setChunkSize(CHUNK_SIZE);
+	}
+
+	@Test
+	public void testIncrementOffset() {
+		assertEquals(0, monitor.getOffset());
+		monitor.incrementOffset();
+		assertEquals(1, monitor.getOffset());
+	}
+
+	@Test
+	public void testResetOffsetManually() {
+		monitor.incrementOffset();
+		monitor.resetOffset();
+		assertEquals(0, monitor.getOffset());
+	}
+
+	@Test
+	public void testResetOffsetAutomatically() {
+		for (int i = 0; i < CHUNK_SIZE; i++) {
+			monitor.incrementOffset();
+		}
+		assertEquals(0, monitor.getOffset());
+	}
+
+	@Test
+	public void testClose() {
+		monitor.incrementOffset();
+		monitor.close();
+		assertTrue(closed);
+		assertEquals(0, monitor.getOffset());
+	}
+
+	@Test
+	public void testOpen() {
+		ExecutionContext executionContext = new ExecutionContext();
+		executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2);
+		monitor.open(executionContext);
+		assertEquals(2, count);
+	}
+
+	@Test
+	public void testOpenWithNullReader() {
+		monitor.setItemReader(null);
+		ExecutionContext executionContext = new ExecutionContext();
+		monitor.open(executionContext);
+		assertEquals(0, monitor.getOffset());
+	}
+
+	@Test(expected = ItemStreamException.class)
+	public void testOpenWithErrorInReader() {
+		monitor.setItemReader(new ItemReader<String>() {
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				throw new IllegalStateException("Expected");
+			}
+		});
+		ExecutionContext executionContext = new ExecutionContext();
+		executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2);
+		monitor.open(executionContext);
+	}
+
+	@Test
+	public void testUpdateOnBoundary() {
+		monitor.resetOffset();
+		ExecutionContext executionContext = new ExecutionContext();
+		monitor.update(executionContext);
+		assertEquals(0, executionContext.size());
+	}
+
+	@Test
+	public void testUpdateVanilla() {
+		monitor.incrementOffset();
+		ExecutionContext executionContext = new ExecutionContext();
+		monitor.update(executionContext);
+		assertEquals(1, executionContext.size());
+	}
+
+}

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

@@ -100,6 +100,8 @@ public class FaultTolerantChunkProcessorTests {
 		catch (RuntimeException e) {
 			assertEquals("Planned failure!", e.getMessage());
 		}
+		processor.process(contribution, chunk);
+		assertEquals(2, chunk.getItems().size());
 		try {
 			processor.process(contribution, chunk);
 			fail();
@@ -107,11 +109,12 @@ public class FaultTolerantChunkProcessorTests {
 		catch (RuntimeException e) {
 			assertEquals("Planned failure!", e.getMessage());
 		}
-		assertEquals(2, chunk.getItems().size());
+		assertEquals(1, chunk.getItems().size());
 		processor.process(contribution, chunk);
+		assertEquals(0, chunk.getItems().size());
 		// foo is written twice because the failure is detected on the second
 		// attempt when throttling
-		assertEquals("[foo, foo, bar]", list.toString());
+		assertEquals("[foo, bar]", list.toString());
 		// but the after listener is only called once, which is important
 		assertEquals(2, after.size());
 	}
@@ -122,7 +125,7 @@ public class FaultTolerantChunkProcessorTests {
 		processor = new FaultTolerantChunkProcessor<String, String>(new PassThroughItemProcessor<String>(),
 				new ItemWriter<String>() {
 					public void write(List<? extends String> items) throws Exception {
-						// Fail is there is more than one item
+						// Fail if there is more than one item
 						if (items.size() > 1) {
 							throw new RuntimeException("Planned failure!");
 						}
@@ -145,6 +148,7 @@ public class FaultTolerantChunkProcessorTests {
 			assertEquals("Planned failure!", e.getMessage());
 		}
 		processor.process(contribution, chunk);
+		processor.process(contribution, chunk);
 
 		assertEquals("[foo, bar]", list.toString());
 		assertEquals("[foo, bar]", after.toString());

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

@@ -42,8 +42,11 @@ import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
 import org.springframework.batch.core.repository.dao.MapStepExecutionDao;
 import org.springframework.batch.core.repository.support.SimpleJobRepository;
 import org.springframework.batch.core.step.AbstractStep;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.retry.policy.MapRetryContextCache;
 import org.springframework.batch.retry.policy.SimpleRetryPolicy;
@@ -72,12 +75,14 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 
 	int count = 0;
 
+	boolean fail = false;
+
 	private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),
 			new MapStepExecutionDao(), new MapExecutionContextDao());
 
 	JobExecution jobExecution;
 
-	private ItemWriter<String> processor = new ItemWriter<String>() {
+	private ItemWriter<String> writer = new ItemWriter<String>() {
 		public void write(List<? extends String> data) throws Exception {
 			processed.addAll(data);
 		}
@@ -98,7 +103,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		factory.setBeanName("step");
 
 		factory.setItemReader(new ListItemReader<String>(new ArrayList<String>()));
-		factory.setItemWriter(processor);
+		factory.setItemWriter(writer);
 		factory.setJobRepository(repository);
 		factory.setTransactionManager(new ResourcelessTransactionManager());
 		factory.setRetryableExceptionClasses(new HashSet<Class<? extends Throwable>>() {
@@ -167,9 +172,70 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		assertEquals(0, stepExecution.getReadSkipCount());
 	}
 
+	@Test
+	public void testRestartAfterFailedWrite() throws Exception {
+
+		factory.setSkipLimit(0);
+		factory.setCommitInterval(3);
+		AbstractItemCountingItemStreamItemReader<String> reader = new AbstractItemCountingItemStreamItemReader<String>() {
+
+			private ItemReader<String> reader;
+
+			@Override
+			protected void doClose() throws Exception {
+				reader = null;
+			}
+
+			@Override
+			protected void doOpen() throws Exception {
+				reader = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f"));
+			}
+
+			@Override
+			protected String doRead() throws Exception {
+				return reader.read();
+			}
+
+		};
+		// Need to set name or else reader will fail to open
+		reader.setName("foo");
+		factory.setItemReader(reader);
+		factory.setStreams(new ItemStream[] { reader });
+		factory.setItemWriter(new ItemWriter<String>() {
+			public void write(List<? extends String> items) throws Exception {
+				if (fail && items.contains("e")) {
+					throw new RuntimeException("Planned failure");
+				}
+				processed.addAll(items);
+			}
+		});
+		factory.setRetryLimit(0);
+		Step step = (Step) factory.getObject();
+
+		fail = true;
+		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
+		repository.add(stepExecution);
+		step.execute(stepExecution);
+
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		assertEquals(4, stepExecution.getWriteCount());
+		assertEquals(6, stepExecution.getReadCount());
+
+		fail = false;
+		ExecutionContext executionContext = stepExecution.getExecutionContext();
+		stepExecution = new StepExecution(step.getName(), jobExecution);
+		stepExecution.setExecutionContext(executionContext);
+		repository.add(stepExecution);
+		step.execute(stepExecution);
+
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+		assertEquals(2, stepExecution.getWriteCount());
+		assertEquals(2, stepExecution.getReadCount());
+	}
+
 	@Test
 	public void testSkipAndRetry() throws Exception {
-		
+
 		factory.setSkipLimit(2);
 		ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("a", "b", "c", "d", "e", "f")) {
 			public String read() {
@@ -198,7 +264,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 	@Test
 	public void testSkipAndRetryWithWriteFailure() throws Exception {
 
-		factory.setListeners(new StepListener[] { new SkipListenerSupport<String,String>() {
+		factory.setListeners(new StepListener[] { new SkipListenerSupport<String, String>() {
 			public void onSkipInWrite(String item, Throwable t) {
 				recovered.add(item);
 				assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
@@ -258,7 +324,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 	public void testSkipAndRetryWithWriteFailureAndNonTrivialCommitInterval() throws Exception {
 
 		factory.setCommitInterval(3);
-		factory.setListeners(new StepListener[] { new SkipListenerSupport<String,String>() {
+		factory.setListeners(new StepListener[] { new SkipListenerSupport<String, String>() {
 			public void onSkipInWrite(String item, Throwable t) {
 				recovered.add(item);
 				assertTrue(TransactionSynchronizationManager.isActualTransactionActive());
@@ -308,16 +374,17 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 
 		// [a, b, c, d, e, f, null]
 		assertEquals(7, provided.size());
-		// [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, a, c, d, e, f, d,
+		// [a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, a, b, c, d, e, f, d,
 		// e, f, d, e, f, d, e, f, d, e, f, d, e, f]
-		assertEquals(37, processed.size());
+		System.err.println(processed);
+		assertEquals(36, processed.size());
 		// [b, d]
 		assertEquals(2, recovered.size());
 	}
 
 	@Test
 	public void testRetryWithNoSkip() throws Exception {
-		
+
 		factory.setRetryLimit(4);
 		factory.setSkipLimit(0);
 		ItemReader<String> provider = new ListItemReader<String>(Arrays.asList("b")) {
@@ -343,8 +410,8 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
 		repository.add(stepExecution);
 		step.execute(stepExecution);
-		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());	
-			
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+
 		List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
 		assertEquals(expectedOutput, written);
 
@@ -440,7 +507,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		repository.add(stepExecution);
 		step.execute(stepExecution);
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		
+
 		List<String> expectedOutput = Arrays.asList(StringUtils.commaDelimitedListToStringArray(""));
 		assertEquals(expectedOutput, written);
 
@@ -489,7 +556,7 @@ public class FaultTolerantStepFactoryBeanRetryTests {
 		repository.add(stepExecution);
 		step.execute(stepExecution);
 		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
-		
+
 		// We added a bogus cache so no items are actually skipped
 		// because they aren't recognised as eligible
 		assertEquals(0, stepExecution.getSkipCount());

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

@@ -30,14 +30,18 @@ import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.SkipListenerSupport;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
+import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemStreamReader;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.ParseException;
 import org.springframework.batch.item.UnexpectedInputException;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
+import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
 import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
 import org.springframework.util.StringUtils;
 
@@ -66,7 +70,11 @@ public class FaultTolerantStepFactoryBeanTests {
 
 	private List<String> processed = new ArrayList<String>();
 
-	protected int count;
+	private int count;
+
+	private boolean opened = false;
+
+	private boolean closed = false;
 
 	private Collection<String> NO_FAILURES = Collections.emptyList();
 
@@ -380,9 +388,10 @@ public class FaultTolerantStepFactoryBeanTests {
 
 		// listeners are called only once chunk is about to commit, so
 		// listener failure does not affect other statistics
-		assertEquals(3, stepExecution.getSkipCount());
 		assertEquals(2, stepExecution.getReadSkipCount());
-		assertEquals(1, stepExecution.getWriteSkipCount());
+		// but we didn't get as far as the write skip in the scan:
+		assertEquals(0, stepExecution.getWriteSkipCount());
+		assertEquals(2, stepExecution.getSkipCount());
 		assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step
 				.getName()));
 	}
@@ -599,10 +608,10 @@ public class FaultTolerantStepFactoryBeanTests {
 		assertEquals(1, stepExecution.getSkipCount());
 		assertEquals(2, stepExecution.getRollbackCount());
 
-		// 1,2,3,4,3,4,3 - two re-processing attempts until the item is
-		// identified and skipped
+		// 1,2,3,4,3,4,4 - two re-processing attempts until the item is
+		// identified and finally skipped on the third attempt
 		assertEquals(7, processed.size());
-		assertEquals("[1, 2, 3, 4, 3, 4, 3]", processed.toString());
+		assertEquals("[1, 2, 3, 4, 3, 4, 4]", processed.toString());
 		assertStepExecutionsAreEqual(stepExecution, repository.getLastStepExecution(jobExecution.getJobInstance(), step
 				.getName()));
 
@@ -683,8 +692,42 @@ public class FaultTolerantStepFactoryBeanTests {
 		}
 	}
 
-	private static class SkipProcessorStub implements ItemProcessor<String, String> {
+	/**
+	 * Check ItemStream is opened
+	 */
+	@Test
+	public void testItemStreamOpenedEvenWithTaskExecutor() throws Exception {
 
+		ItemStreamReader<String> reader = new ItemStreamReader<String>() {
+			public void close() throws ItemStreamException {
+				closed = true;
+			}
+
+			public void open(ExecutionContext executionContext) throws ItemStreamException {
+				opened = true;
+			}
+
+			public void update(ExecutionContext executionContext) throws ItemStreamException {
+			}
+
+			public String read() throws Exception, UnexpectedInputException, ParseException {
+				return null;
+			}
+		};
+
+		factory.setItemReader(reader);
+		factory.setTaskExecutor(new ConcurrentTaskExecutor());
+
+		Step step = (Step) factory.getObject();
+
+		step.execute(stepExecution);
+
+		assertTrue(opened);
+		assertTrue(closed);
+		assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
+	}
+
+	private static class SkipProcessorStub implements ItemProcessor<String, String> {
 		private final Collection<String> failures;
 
 		private boolean rollback = false;

spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java

@@ -126,16 +126,31 @@ public class TaskletStepTests {
 
 	@Test
 	public void testStepExecutor() throws Exception {
-
 		JobExecution jobExecutionContext = new JobExecution(jobInstance);
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecutionContext);
-
 		step.execute(stepExecution);
 		assertEquals(1, processed.size());
 		assertEquals(1, stepExecution.getReadCount());
 		assertEquals(1, stepExecution.getCommitCount());
 	}
 
+	@Test
+	public void testEmptyReader() throws Exception {
+		JobExecution jobExecutionContext = new JobExecution(jobInstance);
+		StepExecution stepExecution = new StepExecution(step.getName(), jobExecutionContext);
+		step = getStep(new String[0]);
+		step.setTasklet(new TestingChunkOrientedTasklet<String>(getReader(new String[0]), itemWriter,
+				new RepeatTemplate()));
+		step.setStepOperations(new RepeatTemplate());
+		step.execute(stepExecution);
+		assertEquals(0, processed.size());
+		assertEquals(0, stepExecution.getReadCount());
+		// Commit after end of data detected (this leads to the commit count
+		// being one greater than people expect if the commit interval is
+		// commensurate with the total number of items).h
+		assertEquals(1, stepExecution.getCommitCount());
+	}
+
 	/**
 	 * StepExecution should be updated after every chunk commit.
 	 */