OPEN - issue BATCH-771: Refactor Listeners for chunk changes

Change step handler to not process items again if already completed

spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java

@@ -134,4 +134,12 @@ public class StepContribution {
 				+ ", writeSkips=" + writeSkipCount + "]";
 	}
 
+	/**
+	 * @param contribution
+	 */
+	public void increment(StepContribution contribution) {
+		itemCount += contribution.getItemCount();
+		readSkipCount += contribution.getReadSkipCount();
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java

@@ -38,7 +38,7 @@ import org.springframework.core.AttributeAccessor;
  * {@link ItemWriter}.
  * 
  * Provides extension points by protected {@link #read(StepContribution)} and
- * {@link #write(Object, StepContribution)} methods that can be overriden to
+ * {@link #write(List, StepContribution)} methods that can be overriden to
  * provide more sophisticated behavior (e.g. skipping).
  * 
  * @author Dave Syer
@@ -46,7 +46,9 @@ import org.springframework.core.AttributeAccessor;
  */
 public class ItemOrientedStepHandler<T, S> implements StepHandler {
 
-	private static final String ITEM_BUFFER_KEY = ItemOrientedStepHandler.class.getName() + ".ITEM_BUFFER_KEY";
+	private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY";
+
+	private static final String OUTPUT_BUFFER_KEY = "OUTPUT_BUFFER_KEY";
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -76,7 +78,7 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 
 	/**
 	 * Get the next item from {@link #read(StepContribution)} and if not null
-	 * pass the item to {@link #write(Object, StepContribution)}. If the
+	 * pass the item to {@link #write(List, StepContribution)}. If the
 	 * {@link ItemProcessor} returns null, the write is omitted and another item
 	 * taken from the reader.
 	 * 
@@ -85,51 +87,53 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 	 */
 	public ExitStatus handle(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
 
-		final List<ReadWrapper<T>> buffer = getItemBuffer(attributes);
+		final List<ItemWrapper<T>> inputs = getInputBuffer(attributes);
+		final List<ItemWrapper<S>> outputs = getOutputBuffer(attributes);
 
 		ExitStatus result = ExitStatus.CONTINUABLE;
 
-		if (buffer.isEmpty()) {
+		if (inputs.isEmpty() && outputs.isEmpty()) {
 
 			result = repeatOperations.iterate(new RepeatCallback() {
 				public ExitStatus doInIteration(final RepeatContext context) throws Exception {
-					ReadWrapper<T> item = read(contribution);
+					ItemWrapper<T> item = read(contribution);
 					contribution.incrementReadSkipCount(item.getSkipCount());
 					if (item.getItem() == null) {
 						return ExitStatus.FINISHED;
 					}
-					buffer.add(item);
+					inputs.add(item);
 					return ExitStatus.CONTINUABLE;
 				}
 			});
 
+			storeInputs(attributes, inputs);
+
 		}
 
-		List<S> processed = new ArrayList<S>();
+		for (Iterator<ItemWrapper<T>> iterator = inputs.iterator(); iterator.hasNext();) {
 
-		for (Iterator<ReadWrapper<T>> iterator = buffer.iterator(); iterator.hasNext();) {
-
-			ReadWrapper<T> item = iterator.next();
+			ItemWrapper<T> item = iterator.next();
 			S output = null;
 
+			// TODO: processor listener
+			output = itemProcessor.process(item.getItem());
+
 			// TODO: segregate read / write / filter count
 			// (this is read count)
 			contribution.incrementItemCount();
-			// TODO: processor listener
-			output = itemProcessor.process(item.getItem());
 
 			// TODO: increment filter count if this is null
 			if (output != null) {
-				processed.add(output);
+				outputs.add(new ItemWrapper<S>(output));
 			}
 
 		}
 
+		storeOutputsAndClearInputs(attributes, outputs, contribution);
+
 		// TODO: use ItemWriter interface properly
 		// TODO: make sure exceptions get handled by the appropriate handler
-		for (S data : processed) {
-			write(data, contribution);
-		}
+		write(outputs, contribution);
 
 		// On successful completion clear the attributes to signal that there is
 		// no more processing
@@ -140,18 +144,65 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 
 	}
 
+	/**
+	 * @param attributes
+	 */
+	private void clearInputs(AttributeAccessor attributes) {
+		attributes.removeAttribute(INPUT_BUFFER_KEY);
+	}
+
+	/**
+	 * @param attributes
+	 * @param inputs
+	 */
+	private void storeInputs(AttributeAccessor attributes, List<ItemWrapper<T>> inputs) {
+		store(attributes, INPUT_BUFFER_KEY, inputs);
+	}
+
+	/**
+	 * Savepoint at end of processing and before writing. The processed items
+	 * ready for output are stored so that if writing fails they can be picked
+	 * up again in the next try. The inputs are finished with so we can clear
+	 * their attribute.
+	 * 
+	 * @param attributes
+	 * @param outputs
+	 */
+	private void storeOutputsAndClearInputs(AttributeAccessor attributes, List<ItemWrapper<S>> outputs,
+			StepContribution contribution) {
+		store(attributes, OUTPUT_BUFFER_KEY, outputs);
+		clearInputs(attributes);
+	}
+
+	/**
+	 * @param attributes
+	 * @param inputBufferKey
+	 * @param outputs
+	 */
+	private <W> void store(AttributeAccessor attributes, String key, W value) {
+		attributes.setAttribute(key, value);
+	}
+
 	private void clearAll(AttributeAccessor attributes) {
 		for (String key : attributes.attributeNames()) {
 			attributes.removeAttribute(key);
 		}
 	}
 
-	private List<ReadWrapper<T>> getItemBuffer(AttributeAccessor attributes) {
-		if (!attributes.hasAttribute(ITEM_BUFFER_KEY)) {
-			attributes.setAttribute(ITEM_BUFFER_KEY, new ArrayList<ReadWrapper<T>>());
+	private List<ItemWrapper<T>> getInputBuffer(AttributeAccessor attributes) {
+		return getBuffer(attributes, INPUT_BUFFER_KEY);
+	}
+
+	private List<ItemWrapper<S>> getOutputBuffer(AttributeAccessor attributes) {
+		return getBuffer(attributes, OUTPUT_BUFFER_KEY);
+	}
+
+	private <W> List<W> getBuffer(AttributeAccessor attributes, String key) {
+		if (!attributes.hasAttribute(key)) {
+			return new ArrayList<W>();
 		}
 		@SuppressWarnings("unchecked")
-		List<ReadWrapper<T>> resource = (List<ReadWrapper<T>>) attributes.getAttribute(ITEM_BUFFER_KEY);
+		List<W> resource = (List<W>) attributes.getAttribute(key);
 		return resource;
 	}
 
@@ -159,8 +210,8 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 	 * @param contribution current context
 	 * @return next item for writing
 	 */
-	protected ReadWrapper<T> read(StepContribution contribution) throws Exception {
-		return new ReadWrapper<T>(doRead());
+	protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
+		return new ItemWrapper<T>(doRead());
 	}
 
 	/**
@@ -173,27 +224,29 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 
 	/**
 	 * 
-	 * @param item the item to write
+	 * @param items the item to write
 	 * @param contribution current context
 	 */
-	protected void write(S item, StepContribution contribution) throws Exception {
-		doWrite(item);
+	protected void write(List<ItemWrapper<S>> items, StepContribution contribution) throws Exception {
+		for (ItemWrapper<S> item : items) {
+			doWrite(item);
+		}
 	}
 
 	/**
 	 * @param item
 	 * @throws Exception
 	 */
-	protected final void doWrite(S item) throws Exception {
+	protected final void doWrite(ItemWrapper<S> item) throws Exception {
 		// TODO: increment write count
-		itemWriter.write(Collections.singletonList(item));
+		itemWriter.write(Collections.singletonList(item.getItem()));
 	}
 
 	/**
 	 * @author Dave Syer
 	 * 
 	 */
-	static protected class ReadWrapper<T> {
+	static protected class ItemWrapper<T> {
 
 		final private T item;
 
@@ -202,7 +255,7 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 		/**
 		 * @param item
 		 */
-		public ReadWrapper(T item) {
+		public ItemWrapper(T item) {
 			this(item, 0);
 		}
 
@@ -210,7 +263,7 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 		 * @param item
 		 * @param skipCount
 		 */
-		public ReadWrapper(T item, int skipCount) {
+		public ItemWrapper(T item, int skipCount) {
 			this.item = item;
 			this.skipCount = skipCount;
 		}
@@ -230,6 +283,16 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
 			return skipCount;
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * 
+		 * @see java.lang.Object#toString()
+		 */
+		@Override
+		public String toString() {
+			return String.format("[%s,%d]", item, skipCount);
+		}
+
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java

@@ -274,8 +274,8 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 						}
 					});
 			StatefulRetryStepHandler<T, S> itemHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
-					getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator, readSkipPolicy,
-					writeSkipPolicy);
+					getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator,
+					readSkipPolicy, writeSkipPolicy);
 			itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners()));
 
 			step.setStepHandler(itemHandler);
@@ -327,9 +327,9 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 		 * @param itemKeyGenerator
 		 */
 		public StatefulRetryStepHandler(ItemReader<? extends T> itemReader,
-				ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter, RepeatOperations chunkOperations, 
-				RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy,
-				ItemSkipPolicy writeSkipPolicy) {
+				ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
+				RepeatOperations chunkOperations, RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator,
+				ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy) {
 			super(itemReader, itemProcessor, itemWriter, chunkOperations);
 			this.retryOperations = retryTemplate;
 			this.itemKeyGenerator = itemKeyGenerator;
@@ -368,13 +368,13 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 		 * count
 		 * @return next item for processing
 		 */
-		protected ReadWrapper<T> read(StepContribution contribution) throws Exception {
-			
+		protected ItemWrapper<T> read(StepContribution contribution) throws Exception {
+
 			int skipCount = 0;
 
 			while (true) {
 				try {
-					return new ReadWrapper<T>(doRead(), skipCount);
+					return new ItemWrapper<T>(doRead(), skipCount);
 				}
 				catch (Exception e) {
 					try {
@@ -410,37 +410,48 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
 		/**
 		 * Execute the business logic, delegating to the writer.<br/>
 		 * 
-		 * Process the item with the {@link ItemWriter} in a stateful retry. Any
+		 * Process the items with the {@link ItemWriter} in a stateful retry. Any
 		 * {@link SkipListener} provided is called when retry attempts are
 		 * exhausted. The listener callback (on write failure) will happen in
 		 * the next transaction automatically.<br/>
 		 */
 		@Override
-		protected void write(final S item, final StepContribution contribution) throws Exception {
-			RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
-				public Object doWithRetry(RetryContext context) throws Throwable {
-					doWrite(item);
-					return null;
-				}
-			}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item) : item);
-			retryCallback.setRecoveryCallback(new RecoveryCallback() {
-				public Object recover(RetryContext context) {
-					Throwable t = context.getLastThrowable();
-					if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
-						contribution.incrementWriteSkipCount();
-						try {
-							listener.onSkipInWrite(item, t);
-						}
-						catch (RuntimeException ex) {
-							throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
-						}
-					} else {
-						throw new RetryException("Non-skippable exception in recoverer", t);
+		protected void write(final List<ItemWrapper<S>> items, final StepContribution contribution) throws Exception {
+			
+			for (final ItemWrapper<S> item : new ArrayList<ItemWrapper<S>>(items)) {
+
+				RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {
+					public Object doWithRetry(RetryContext context) throws Throwable {
+						doWrite(item);
+						return null;
 					}
-					return null;
-				}
-			});
-			retryOperations.execute(retryCallback);
+				}, itemKeyGenerator != null ? itemKeyGenerator.getKey(item.getItem()) : item);
+
+				retryCallback.setRecoveryCallback(new RecoveryCallback() {
+
+					public Object recover(RetryContext context) {
+						Throwable t = context.getLastThrowable();
+						if (writeSkipPolicy.shouldSkip(t, contribution.getStepSkipCount())) {
+							contribution.incrementWriteSkipCount();
+							items.remove(item);
+							try {
+								listener.onSkipInWrite(item.getItem(), t);
+							}
+							catch (RuntimeException ex) {
+								throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, t);
+							}
+						}
+						else {
+							throw new RetryException("Non-skippable exception in recoverer", t);
+						}
+						return null;
+
+					}
+				});
+
+				retryOperations.execute(retryCallback);
+
+			}
 		}
 	}
 

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java

@@ -233,8 +233,7 @@ public class StepHandlerStep extends AbstractStep {
 
 				AttributeAccessor attributes = attributeQueue.poll();
 				if (attributes == null) {
-					attributes = new AttributeAccessorSupport() {
-					};
+					attributes = new BasicAttributeAccessor();
 				}
 
 				try {
@@ -258,6 +257,11 @@ public class StepHandlerStep extends AbstractStep {
 						if (attributes.attributeNames().length > 0) {
 							attributeQueue.add(attributes);
 						}
+
+						// Apply the contribution to the step
+						// even if unsuccessful
+						stepExecution.apply(contribution);
+
 					}
 
 					contribution.incrementCommitCount();
@@ -274,10 +278,6 @@ public class StepHandlerStep extends AbstractStep {
 						Thread.currentThread().interrupt();
 					}
 
-					// Apply the contribution to the step
-					// only if chunk was successful
-					stepExecution.apply(contribution);
-
 					try {
 						stream.update(stepExecution.getExecutionContext());
 					}
@@ -322,11 +322,11 @@ public class StepHandlerStep extends AbstractStep {
 
 				}
 				catch (Error e) {
-					processRollback(stepExecution, contribution, fatalException, transaction);
+					processRollback(stepExecution, fatalException, transaction);
 					throw e;
 				}
 				catch (Exception e) {
-					processRollback(stepExecution, contribution, fatalException, transaction);
+					processRollback(stepExecution, fatalException, transaction);
 					throw e;
 				}
 				finally {
@@ -351,15 +351,12 @@ public class StepHandlerStep extends AbstractStep {
 
 	/**
 	 * @param stepExecution
-	 * @param contribution
 	 * @param fatalException
 	 * @param transaction
 	 */
-	private void processRollback(final StepExecution stepExecution, final StepContribution contribution,
-			final ExceptionHolder fatalException, TransactionStatus transaction) {
+	private void processRollback(final StepExecution stepExecution, final ExceptionHolder fatalException,
+			TransactionStatus transaction) {
 
-		stepExecution.incrementReadSkipCountBy(contribution.getReadSkipCount());
-		stepExecution.incrementWriteSkipCountBy(contribution.getWriteSkipCount());
 		/*
 		 * Any exception thrown within the transaction should automatically
 		 * cause the transaction to rollback.
@@ -387,6 +384,13 @@ public class StepHandlerStep extends AbstractStep {
 		}
 	}
 
+	/**
+	 * @author Dave Syer
+	 *
+	 */
+	private static final class BasicAttributeAccessor extends AttributeAccessorSupport {
+	}
+
 	private static class ExceptionHolder {
 
 		private Exception exception;

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

@@ -343,14 +343,13 @@ public class SkipLimitStepFactoryBeanTests {
 		factory.setSkipLimit(4);
 		factory.setItemReader(reader);
 		factory.setItemWriter(writer);
+		factory.setCommitInterval(3); // includes all expected skips
 
 		Step step = (Step) factory.getObject();
 
 		StepExecution stepExecution = jobExecution.createStepExecution(step);
 
 		step.execute(stepExecution);
-		System.err.println(writer.written);
-		System.err.println(reader.processed);
 		assertEquals(4, stepExecution.getSkipCount());
 		assertEquals(2, stepExecution.getReadSkipCount());
 		assertEquals(2, stepExecution.getWriteSkipCount());

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

@@ -43,6 +43,7 @@ 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.core.step.skip.SkipLimitExceededException;
+import org.springframework.batch.item.ItemKeyGenerator;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.ListItemReader;
@@ -283,17 +284,17 @@ public class StatefulRetryStepFactoryBeanTests {
 		assertEquals(4, processed.size());
 		// []
 		assertEquals(0, recovered.size());
-		assertEquals(0, stepExecution.getItemCount());
+		assertEquals(1, stepExecution.getItemCount());
 	}
 
 	@Test
 	public void testNonSkippableException() throws Exception {
-		
+
 		// Very specific skippable exception
 		factory.setSkippableExceptionClasses(new Class[] { UnsupportedOperationException.class });
 		// ...which is not retryable...
 		factory.setRetryableExceptionClasses(new Class<?>[0]);
-		
+
 		factory.setSkipLimit(1);
 		List<String> items = Arrays.asList(new String[] { "b" });
 		ItemReader<String> provider = new ListItemReader<String>(items) {
@@ -323,7 +324,7 @@ public class StatefulRetryStepFactoryBeanTests {
 		catch (RuntimeException e) {
 			// expected
 			String message = e.getMessage();
-			assertTrue("Wrong message: "+message, message.contains("Write error - planned but not skippable."));
+			assertTrue("Wrong message: " + message, message.contains("Write error - planned but not skippable."));
 		}
 
 		assertEquals(0, stepExecution.getSkipCount());
@@ -333,7 +334,7 @@ public class StatefulRetryStepFactoryBeanTests {
 		assertEquals(1, processed.size());
 		// []
 		assertEquals(0, recovered.size());
-		assertEquals(0, stepExecution.getItemCount());
+		assertEquals(1, stepExecution.getItemCount());
 	}
 
 	@Test
@@ -376,7 +377,7 @@ public class StatefulRetryStepFactoryBeanTests {
 		assertEquals(4, processed.size());
 		// []
 		assertEquals(0, recovered.size());
-		assertEquals(0, stepExecution.getItemCount());
+		assertEquals(1, stepExecution.getItemCount());
 	}
 
 	@Test
@@ -390,7 +391,7 @@ public class StatefulRetryStepFactoryBeanTests {
 		factory.setCacheCapacity(2);
 		ItemReader<String> provider = new ItemReader<String>() {
 			public String read() {
-				String item = ""+count;
+				String item = "" + count;
 				provided.add(item);
 				count++;
 				if (count >= 10) {
@@ -409,6 +410,12 @@ public class StatefulRetryStepFactoryBeanTests {
 		};
 		factory.setItemReader(provider);
 		factory.setItemWriter(itemWriter);
+		factory.setItemKeyGenerator(new ItemKeyGenerator() {
+			public Object getKey(Object item) {
+				// return random object so the cache fills up
+				return new Object();
+			}
+		});
 		AbstractStep step = (AbstractStep) factory.getObject();
 
 		StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
@@ -420,13 +427,14 @@ public class StatefulRetryStepFactoryBeanTests {
 			// expected
 		}
 
-		assertEquals(1, stepExecution.getSkipCount());
+		// We added a bogus key generator so no items are actually skipped
+		// because they aren't recognised as eligible
+		assertEquals(0, stepExecution.getSkipCount());
 		// only one item processed but three (the commit interval) were provided
 		// [0, 1, 2]
 		assertEquals(3, provided.size());
-		// TODO: this is a bug: 0 was skipped but it came back in the buffer for the second try
-		// [0, 0, 1, 0, 0]
-		assertEquals(5, processed.size());
+		// [0, 0, 0]
+		assertEquals(3, processed.size());
 		// []
 		assertEquals(0, recovered.size());
 	}