Add ProcessErrorListener for AsyncItemProcessor to allow notification of async processing failures

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2015 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;
+
+import org.springframework.batch.item.ItemProcessor;
+
+/**
+ * Interface to mark {@link ItemProcessor} which support injection of a {@link ItemProcessListener}.
+ * 
+ * @author Dominik Bartholdi
+ *
+ */
+public interface ItemProcessListenerSupport<I, O> {
+
+	public void setItemProcessListener(ItemProcessListener<I, O> itemProcessListener);
+
+}

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

@@ -18,6 +18,8 @@
 
 import java.util.List;
 
+import org.springframework.batch.core.ItemProcessListenerSupport;
+import org.springframework.batch.core.ItemProcessListener;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.MulticasterBatchListener;
@@ -99,6 +101,9 @@ public void setListeners(List<? extends StepListener> listeners) {
 	 */
 	public void registerListener(StepListener listener) {
 		this.listener.register(listener);
+		if((itemProcessor instanceof ItemProcessListenerSupport) && (listener instanceof ItemProcessListener)){
+			((ItemProcessListenerSupport)itemProcessor).setItemProcessListener((ItemProcessListener)listener);
+		}
 	}
 
 	/**

spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java

@@ -19,6 +19,8 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.FutureTask;
 
+import org.springframework.batch.core.ItemProcessListenerSupport;
+import org.springframework.batch.core.ItemProcessListener;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.scope.context.StepContext;
 import org.springframework.batch.core.scope.context.StepSynchronizationManager;
@@ -48,11 +50,13 @@
  * @param <O> the output object type (will be wrapped in a Future)
  * @see AsyncItemWriter
  */
-public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {
+public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean, ItemProcessListenerSupport<I, O> {
 
 	private ItemProcessor<I, O> delegate;
 
 	private TaskExecutor taskExecutor = new SyncTaskExecutor();
+
+	private ItemProcessListener<I, O> itemProcessListener;
 
 	/**
 	 * Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
@@ -72,7 +76,7 @@ public void afterPropertiesSet() throws Exception {
 	public void setDelegate(ItemProcessor<I, O> delegate) {
 		this.delegate = delegate;
 	}
-
+	
 	/**
 	 * The {@link TaskExecutor} to use to allow the item processing to proceed
 	 * in the background. Defaults to a {@link SyncTaskExecutor} so no threads
@@ -83,7 +87,16 @@ public void setDelegate(ItemProcessor<I, O> delegate) {
 	public void setTaskExecutor(TaskExecutor taskExecutor) {
 		this.taskExecutor = taskExecutor;
 	}
-
+
+	/**
+	 * The {@link ItemProcessListener} to be called in case of an exception.
+	 * 
+	 * @param itemProcessListener a {@link ItemProcessListener} to be notified about exceptions.
+	 */
+	public void setItemProcessListener(ItemProcessListener<I, O> itemProcessListener) {
+		this.itemProcessListener = itemProcessListener;
+	}
+
 	/**
 	 * Transform the input by delegating to the provided item processor. The
 	 * return value is wrapped in a {@link Future} so that clients can unpack it
@@ -101,6 +114,12 @@ public O call() throws Exception {
 				try {
 					return delegate.process(item);
 				}
+				catch (Exception e) {
+					if (itemProcessListener != null) {
+						itemProcessListener.onProcessError(item, e);
+					}
+					throw e;
+				}
 				finally {
 					if (stepExecution != null) {
 						StepSynchronizationManager.close();