OPEN - issue BATCH-498: Skip method should get ExecutionContext as an argument

Added SkipListener as new StepListener.

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

@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+/**
+ * Interface for listener to skipped items. Callbacks will be called by
+ * {@link Step} implementations at the appropriate time in the step lifecycle.
+ * Callbacks must always be made (where relevant) in a transaction that is still
+ * valid: i.e. possibly on a read error, but not on a write error.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public interface SkipListener extends StepListener {
+
+	/**
+	 * Callback for a failure on read that is legal, so is not going to be
+	 * re-thrown.
+	 * 
+	 * @param t
+	 */
+	void onSkipInRead(Throwable t);
+
+	/**
+	 * This item failed on write with the given exception, and a skip was called
+	 * for. The callback is deferred until a new transaction is available. This
+	 * callback might occur more than once for the same item, but only once in
+	 * successful transaction.
+	 * 
+	 * 
+	 * @param item the failed item
+	 * @param t the cause of the failure
+	 */
+	void onSkipInWrite(Object item, Throwable t);
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java

@@ -50,23 +50,24 @@ public class SimpleJob extends AbstractJob {
 	private CompositeExecutionJobListener listener = new CompositeExecutionJobListener();
 
 	/**
-	 * Public setter for injecting {@link JobExecutionListener}s. They will all be given
-	 * the {@link JobExecutionListener} callbacks at the appropriate point in the job.
+	 * Public setter for injecting {@link JobExecutionListener}s. They will all
+	 * be given the listener callbacks at the appropriate point in the job.
 	 * 
 	 * @param listeners the listeners to set.
 	 */
-	public void setJobListeners(JobExecutionListener[] listeners) {
+	public void setJobExecutionListeners(JobExecutionListener[] listeners) {
 		for (int i = 0; i < listeners.length; i++) {
 			this.listener.register(listeners[i]);
 		}
 	}
 
 	/**
-	 * Register a single listener for the {@link JobExecutionListener} callbacks.
+	 * Register a single listener for the {@link JobExecutionListener}
+	 * callbacks.
 	 * 
 	 * @param listener a {@link JobExecutionListener}
 	 */
-	public void registerListener(JobExecutionListener listener) {
+	public void registerJobExecutionListener(JobExecutionListener listener) {
 		this.listener.register(listener);
 	}
 
@@ -75,7 +76,8 @@ public class SimpleJob extends AbstractJob {
 	 * {@link Step}.
 	 * 
 	 * @see org.springframework.batch.core.Job#execute(org.springframework.batch.core.JobExecution)
-	 * @throws StartLimitExceededException if start limit of one of the steps was exceeded
+	 * @throws StartLimitExceededException if start limit of one of the steps
+	 * was exceeded
 	 */
 	public void execute(JobExecution execution) throws JobExecutionException {
 
@@ -110,8 +112,8 @@ public class SimpleJob extends AbstractJob {
 
 					StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step);
 
-					boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0
-							&& !lastStepExecution.getExitStatus().equals(ExitStatus.FINISHED)) ? true : false;
+					boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution
+							.getExitStatus().equals(ExitStatus.FINISHED)) ? true : false;
 
 					if (isRestart && lastStepExecution != null) {
 						currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());

spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeSkipListener.java

@@ -0,0 +1,88 @@
+/*
+ * 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.listener;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.repeat.ExitStatus;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class CompositeSkipListener implements SkipListener {
+
+	private List listeners = new ArrayList();
+
+	/**
+	 * Public setter for the listeners.
+	 * 
+	 * @param listeners
+	 */
+	public void setListeners(SkipListener[] listeners) {
+		this.listeners = Arrays.asList(listeners);
+	}
+
+	/**
+	 * Register additional listener.
+	 * 
+	 * @param stepExecutionListener
+	 */
+	public void register(SkipListener listener) {
+		if (!listeners.contains(listener)) {
+			listeners.add(listener);
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.domain.StepListener#onError(java.lang.Throwable)
+	 */
+	public ExitStatus onErrorInStep(StepExecution stepExecution, Throwable e) {
+		ExitStatus status = null;
+		for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
+			StepExecutionListener listener = (StepExecutionListener) iterator.next();
+			ExitStatus close = listener.onErrorInStep(stepExecution, e);
+			status = status!=null ? status.and(close): close;
+		}
+		return status;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.SkipListener#onSkipInRead(java.lang.Throwable)
+	 */
+	public void onSkipInRead(Throwable t) {
+		for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
+			SkipListener listener = (SkipListener) iterator.next();
+			listener.onSkipInRead(t);
+		}
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.core.SkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable)
+	 */
+	public void onSkipInWrite(Object item, Throwable t) {
+		for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
+			SkipListener listener = (SkipListener) iterator.next();
+			listener.onSkipInWrite(item, t);
+		}
+	}
+}

spring-batch-core/src/main/java/org/springframework/batch/core/listener/MulticasterBatchListener.java

@@ -15,12 +15,13 @@
  */
 package org.springframework.batch.core.listener;
 
-import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.ChunkListener;
 import org.springframework.batch.core.ItemReadListener;
 import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.SkipListener;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.StepListener;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.repeat.ExitStatus;
 
@@ -29,7 +30,7 @@ import org.springframework.batch.repeat.ExitStatus;
  * 
  */
 public class MulticasterBatchListener implements StepExecutionListener, ChunkListener, ItemReadListener,
-		ItemWriteListener {
+		ItemWriteListener, SkipListener {
 
 	private CompositeStepExecutionListener stepListener = new CompositeStepExecutionListener();
 
@@ -39,6 +40,8 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis
 
 	private CompositeItemWriteListener itemWriteListener = new CompositeItemWriteListener();
 
+	private CompositeSkipListener skipListener = new CompositeSkipListener();
+
 	/**
 	 * Initialise the listener instance.
 	 */
@@ -77,6 +80,9 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis
 		if (listener instanceof ItemWriteListener) {
 			this.itemWriteListener.register((ItemWriteListener) listener);
 		}
+		if (listener instanceof SkipListener) {
+			this.skipListener.register((SkipListener) listener);
+		}
 	}
 
 	/**
@@ -170,4 +176,21 @@ public class MulticasterBatchListener implements StepExecutionListener, ChunkLis
 		itemWriteListener.onWriteError(ex, item);
 	}
 
+	/**
+	 * @param t
+	 * @see org.springframework.batch.core.listener.CompositeSkipListener#onSkipInRead(java.lang.Throwable)
+	 */
+	public void onSkipInRead(Throwable t) {
+		skipListener.onSkipInRead(t);
+	}
+
+	/**
+	 * @param item
+	 * @param t
+	 * @see org.springframework.batch.core.listener.CompositeSkipListener#onSkipInWrite(java.lang.Object, java.lang.Throwable)
+	 */
+	public void onSkipInWrite(Object item, Throwable t) {
+		skipListener.onSkipInWrite(item, t);
+	}
+
 }

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

@@ -18,11 +18,12 @@ package org.springframework.batch.core.step.item;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.ChunkListener;
 import org.springframework.batch.core.ItemReadListener;
 import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.batch.core.SkipListener;
 import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.CompositeChunkListener;
 import org.springframework.batch.core.listener.CompositeItemReadListener;
 import org.springframework.batch.core.listener.CompositeItemWriteListener;
@@ -169,4 +170,19 @@ class BatchListenerFactoryHelper {
 		return (StepExecutionListener[]) list.toArray(new StepExecutionListener[list.size()]);
 	}
 
+	/**
+	 * @param listeners
+	 * @return
+	 */
+	public SkipListener[] getSkipListeners(StepListener[] listeners) {
+		List list = new ArrayList();
+		for (int i = 0; i < listeners.length; i++) {
+			StepListener listener = listeners[i];
+			if (listener instanceof SkipListener) {
+				list.add(listener);
+			}
+		}
+		return (SkipListener[]) list.toArray(new SkipListener[list.size()]);
+	}
+
 }

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

@@ -168,9 +168,9 @@ public class ItemOrientedStep extends AbstractStep {
 	 * 
 	 * @param listeners an array of listener objects of known types.
 	 */
-	public void setStepListeners(StepExecutionListener[] listeners) {
+	public void setStepExecutionListeners(StepExecutionListener[] listeners) {
 		for (int i = 0; i < listeners.length; i++) {
-			registerStepListener(listeners[i]);
+			registerStepExecutionListener(listeners[i]);
 		}
 	}
 
@@ -180,7 +180,7 @@ public class ItemOrientedStep extends AbstractStep {
 	 * 
 	 * @param listener a {@link StepExecutionListener}
 	 */
-	public void registerStepListener(StepExecutionListener listener) {
+	public void registerStepExecutionListener(StepExecutionListener listener) {
 		this.listener.register(listener);
 	}
 

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

@@ -15,12 +15,14 @@
  */
 package org.springframework.batch.core.step.item;
 
-import java.util.HashSet;
-import java.util.Set;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.SkipListener;
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.listener.CompositeSkipListener;
 import org.springframework.batch.core.step.skip.ItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
 import org.springframework.batch.item.ItemKeyGenerator;
@@ -44,13 +46,39 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 
 	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
 
-	private ItemKeyGenerator itemKeyGenerator = new ItemKeyGenerator() {
+	private ItemKeyGenerator defaultItemKeyGenerator = new ItemKeyGenerator() {
 		public Object getKey(Object item) {
 			return item;
 		}
 	};
 
-	private Set skippedItems = new HashSet();
+	private ItemKeyGenerator itemKeyGenerator = defaultItemKeyGenerator;
+
+	private CompositeSkipListener listener = new CompositeSkipListener();
+
+	private Map skippedExceptions = new HashMap();
+
+	/**
+	 * Register some {@link SkipListener}s with the handler. Each will get the
+	 * callbacks in the order specified at the correct stage if a skip occurs.
+	 * 
+	 * @param listeners
+	 */
+	public void setSkipListeners(SkipListener[] listeners) {
+		for (int i = 0; i < listeners.length; i++) {
+			registerSkipListener(listeners[i]);
+		}
+	}
+
+	/**
+	 * Register a listener for callbacks at the appropriate stages in a skip
+	 * process.
+	 * 
+	 * @param listener a {@link SkipListener}
+	 */
+	public void registerSkipListener(SkipListener listener) {
+		this.listener.register(listener);
+	}
 
 	/**
 	 * Public setter for the {@link ItemKeyGenerator}. Defaults to just return
@@ -60,9 +88,13 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 	 * reader does any buffering the key generator might need to take care to
 	 * only use data that do not change on write).
 	 * 
-	 * @param itemKeyGenerator the itemKeyGenerator to set
+	 * @param itemKeyGenerator the {@link ItemKeyGenerator} to set. If null
+	 * resets to default value.
 	 */
 	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
+		if (itemKeyGenerator == null) {
+			itemKeyGenerator = defaultItemKeyGenerator;
+		}
 		this.itemKeyGenerator = itemKeyGenerator;
 	}
 
@@ -95,9 +127,14 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 			try {
 
 				Object item = doRead();
-				while (item != null && skippedItems.contains(itemKeyGenerator.getKey(item))) {
-					logger.debug("Skipping item on input: " + item);
+				Object key = itemKeyGenerator.getKey(item);
+				while (item != null && skippedExceptions.containsKey(key)) {
+					logger.debug("Skipping item on input, previously failed on output; key=[" + key + "]");
+					if (listener != null) {
+						listener.onSkipInWrite(item, (Throwable) skippedExceptions.get(key));
+					}
 					item = doRead();
+					key = itemKeyGenerator.getKey(item);
 				}
 				return item;
 
@@ -106,6 +143,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 				if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
 					// increment skip count and try again
 					contribution.incrementSkipCount();
+					if (listener != null) {
+						listener.onSkipInRead(e);
+					}
+					logger.debug("Skipping failed input", e);
 				}
 				else {
 					// re-throw only when the skip policy runs out of patience
@@ -135,7 +176,10 @@ public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
 		catch (Exception e) {
 			if (itemSkipPolicy.shouldSkip(e, contribution.getStepSkipCount())) {
 				contribution.incrementSkipCount();
-				skippedItems.add(key);
+				// don't call the listener here - the transaction is going to
+				// roll back
+				skippedExceptions.put(key, e);
+				logger.debug("Added item to skip list; key=" + key);
 			}
 			// always re-throw exception on write
 			throw e;

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

@@ -108,13 +108,13 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
 			step.registerStream((ItemStream) itemReader);
 		}
 		if (itemReader instanceof StepExecutionListener) {
-			step.registerStepListener((StepExecutionListener) itemReader);
+			step.registerStepExecutionListener((StepExecutionListener) itemReader);
 		}
 		if (itemWriter instanceof ItemStream) {
 			step.registerStream((ItemStream) itemWriter);
 		}
 		if (itemWriter instanceof StepExecutionListener) {
-			step.registerStepListener((StepExecutionListener) itemWriter);
+			step.registerStepExecutionListener((StepExecutionListener) itemWriter);
 		}
 
 		BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
@@ -128,7 +128,7 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
 		setItemReader(itemReader);
 		setItemWriter(itemWriter);
 
-		step.setStepListeners(stepListeners);
+		step.setStepExecutionListeners(stepListeners);
 		step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
 
 		step.setChunkOperations(chunkOperations);

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

@@ -15,10 +15,9 @@
  */
 package org.springframework.batch.core.step.item;
 
-import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepExecutionListener;
-import org.springframework.batch.core.listener.MulticasterBatchListener;
+import org.springframework.batch.core.StepListener;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemWriter;
@@ -49,8 +48,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 
 	private StepListener[] listeners = new StepListener[0];
 
-	private MulticasterBatchListener listener = new MulticasterBatchListener();
-
 	private TaskExecutor taskExecutor;
 
 	private ItemHandler itemHandler;
@@ -89,6 +86,14 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 	public void setListeners(StepListener[] listeners) {
 		this.listeners = listeners;
 	}
+	
+	/**
+	 * Protected getter for the {@link StepListener}s.
+	 * @return the listeners
+	 */
+	protected StepListener[] getListeners() {
+		return listeners;
+	}
 
 	/**
 	 * Protected getter for the step operations to make them available in
@@ -151,16 +156,6 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 
 		step.setStreams(streams);
 
-		for (int i = 0; i < listeners.length; i++) {
-			StepListener listener = listeners[i];
-			if (listener instanceof StepExecutionListener) {
-				step.registerStepListener((StepExecutionListener) listener);
-			}
-			else {
-				this.listener.register(listener);
-			}
-		}
-
 		ItemReader itemReader = getItemReader();
 		ItemWriter itemWriter = getItemWriter();
 
@@ -171,13 +166,13 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 			step.registerStream((ItemStream) itemReader);
 		}
 		if (itemReader instanceof StepExecutionListener) {
-			step.registerStepListener((StepExecutionListener) itemReader);
+			step.registerStepExecutionListener((StepExecutionListener) itemReader);
 		}
 		if (itemWriter instanceof ItemStream) {
 			step.registerStream((ItemStream) itemWriter);
 		}
 		if (itemWriter instanceof StepExecutionListener) {
-			step.registerStepListener((StepExecutionListener) itemWriter);
+			step.registerStepExecutionListener((StepExecutionListener) itemWriter);
 		}
 
 		BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
@@ -197,7 +192,7 @@ public class SimpleStepFactoryBean extends AbstractStepFactoryBean {
 		setItemReader(itemReader);
 		setItemWriter(itemWriter);
 
-		step.setStepListeners(stepListeners);
+		step.setStepExecutionListeners(stepListeners);
 
 		stepOperations = new RepeatTemplate();
 

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

@@ -4,6 +4,7 @@ import java.util.Arrays;
 
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
+import org.springframework.batch.item.ItemKeyGenerator;
 import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
 
 /**
@@ -29,6 +30,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 
 	private Class[] fatalExceptionClasses = new Class[] { Error.class };
 
+	private ItemKeyGenerator itemKeyGenerator;
+
 	/**
 	 * Public setter for a limit that determines skip policy. If this value is
 	 * positive then an exception in chunk processing will cause the item to be
@@ -62,6 +65,17 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 		this.fatalExceptionClasses = fatalExceptionClasses;
 	}
 
+	/**
+	 * Public setter for the {@link ItemKeyGenerator}. This is used to identify
+	 * failed items so they can be skipped if encountered again, generally in
+	 * another transaction.
+	 * 
+	 * @param itemKeyGenerator the {@link ItemKeyGenerator} to set.
+	 */
+	public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
+		this.itemKeyGenerator = itemKeyGenerator;
+	}
+
 	/**
 	 * Uses the {@link #skipLimit} value to configure item handler and and
 	 * exception handler.
@@ -72,6 +86,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 		ItemSkipPolicyItemHandler itemHandler = new ItemSkipPolicyItemHandler(getItemReader(), getItemWriter());
 
 		if (skipLimit > 0) {
+
 			/*
 			 * If there is a skip limit (not the default) then we are prepared
 			 * to absorb exceptions at the step level because the failed items
@@ -85,6 +100,11 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean {
 			exceptionHandler.setFatalExceptionClasses(fatalExceptionClasses);
 			setExceptionHandler(exceptionHandler);
 			getStepOperations().setExceptionHandler(getExceptionHandler());
+			itemHandler.setItemKeyGenerator(itemKeyGenerator);
+
+			BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
+			itemHandler.setSkipListeners(helper.getSkipListeners(getListeners()));
+
 		}
 		else {
 			// This is the default in ItemOrientedStep anyway...