Add null check to AsyncItemWriter

The AsyncItemWriter is used in conjunction with the AsyncItemProcessor
to unwrap the Futures that are returned by that processor.
Traditionally when an ItemProcessor returns null, it's considered having
been filtered out and should not be passed to the ItemWriter.  In this
case, the AsyncItemWriter was not checking for nulls so they were being
passed to the delegate ItemWriter.  Most of the OOTB ItemWriters do not
perform a null check prior to doing the write so they were throwing NPEs
when using this paradigm.

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

@@ -15,18 +15,18 @@
  */
 package org.springframework.batch.integration.async;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Future;
-
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Future;
+
 public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBean {
 	
 	private ItemWriter<T> delegate;
-	
+
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
 	}
@@ -38,12 +38,23 @@ public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBe
 		this.delegate = delegate;
 	}
 
+	/**
+	 * In the processing of the {@link java.util.concurrent.Future}s passed, nulls are <em>not</em> passed to the
+	 * delegate since they are considered filtered out by the {@link org.springframework.batch.integration.async.AsyncItemProcessor}'s
+	 * delegated {@link org.springframework.batch.item.ItemProcessor}.
+	 *
+	 * @param items {@link java.util.concurrent.Future}s to be upwrapped and passed to the delegate
+	 * @throws Exception
+	 */
 	public void write(List<? extends Future<T>> items) throws Exception {
 		List<T> list = new ArrayList<T>();
 		for (Future<T> future : items) {
-			list.add(future.get());
+			T item = future.get();
+
+			if(item != null) {
+				list.add(future.get());
+			}
 		}
 		delegate.write(list);
 	}
-
 }

spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2014 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.integration.async;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * @author mminella
+ */
+public class AsyncItemWriterTests {
+
+	private AsyncItemWriter<String> writer;
+	private List<String> writtenItems;
+	private TaskExecutor taskExecutor;
+
+	@Before
+	public void setup() {
+		taskExecutor = new SimpleAsyncTaskExecutor();
+		writtenItems = new ArrayList<String>();
+		writer = new AsyncItemWriter<String>();
+		writer.setDelegate(new ListItemWriter(writtenItems));
+	}
+
+	@Test
+	public void testRoseyScenario() throws Exception {
+		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
+
+		processedItems.add(new FutureTask<String>(new Callable<String>() {
+			@Override
+			public String call() throws Exception {
+				return "foo";
+			}
+		}));
+
+		processedItems.add(new FutureTask<String>(new Callable<String>() {
+			@Override
+			public String call() throws Exception {
+				return "bar";
+			}
+		}));
+
+		for (FutureTask<String> processedItem : processedItems) {
+			taskExecutor.execute(processedItem);
+		}
+
+		writer.write(processedItems);
+
+		assertEquals(2, writtenItems.size());
+		assertTrue(writtenItems.contains("foo"));
+		assertTrue(writtenItems.contains("bar"));
+	}
+
+	@Test
+	public void testFilteredItem() throws Exception {
+		List<FutureTask<String>> processedItems = new ArrayList<FutureTask<String>>();
+
+		processedItems.add(new FutureTask<String>(new Callable<String>() {
+			@Override
+			public String call() throws Exception {
+				return "foo";
+			}
+		}));
+
+		processedItems.add(new FutureTask<String>(new Callable<String>() {
+			@Override
+			public String call() throws Exception {
+				return null;
+			}
+		}));
+
+		for (FutureTask<String> processedItem : processedItems) {
+			taskExecutor.execute(processedItem);
+		}
+
+		writer.write(processedItems);
+
+		assertEquals(1, writtenItems.size());
+		assertTrue(writtenItems.contains("foo"));
+	}
+
+	private class ListItemWriter implements ItemWriter<String> {
+
+		protected List<String> items;
+
+		public ListItemWriter(List<String> items) {
+			this.items = items;
+		}
+
+		@Override
+		public void write(List<? extends String> items) throws Exception {
+			this.items.addAll(items);
+		}
+	}
+}