From 2fec70d25eb5f04eca28542327f08d589ce1594b Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Thu, 11 Sep 2014 11:50:07 -0500 Subject: [PATCH] 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. --- .../integration/async/AsyncItemWriter.java | 25 ++-- .../async/AsyncItemWriterTests.java | 119 ++++++++++++++++++ 2 files changed, 137 insertions(+), 7 deletions(-) create mode 100644 spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java index 0bf837ba6..fdfa0fef4 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java +++ b/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 implements ItemWriter>, InitializingBean { private ItemWriter delegate; - + public void afterPropertiesSet() throws Exception { Assert.notNull(delegate, "A delegate ItemWriter must be provided."); } @@ -38,12 +38,23 @@ public class AsyncItemWriter implements ItemWriter>, InitializingBe this.delegate = delegate; } + /** + * In the processing of the {@link java.util.concurrent.Future}s passed, nulls are not 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> items) throws Exception { List list = new ArrayList(); for (Future future : items) { - list.add(future.get()); + T item = future.get(); + + if(item != null) { + list.add(future.get()); + } } delegate.write(list); } - } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java new file mode 100644 index 000000000..3fd6b63a3 --- /dev/null +++ b/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 writer; + private List writtenItems; + private TaskExecutor taskExecutor; + + @Before + public void setup() { + taskExecutor = new SimpleAsyncTaskExecutor(); + writtenItems = new ArrayList(); + writer = new AsyncItemWriter(); + writer.setDelegate(new ListItemWriter(writtenItems)); + } + + @Test + public void testRoseyScenario() throws Exception { + List> processedItems = new ArrayList>(); + + processedItems.add(new FutureTask(new Callable() { + @Override + public String call() throws Exception { + return "foo"; + } + })); + + processedItems.add(new FutureTask(new Callable() { + @Override + public String call() throws Exception { + return "bar"; + } + })); + + for (FutureTask 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> processedItems = new ArrayList>(); + + processedItems.add(new FutureTask(new Callable() { + @Override + public String call() throws Exception { + return "foo"; + } + })); + + processedItems.add(new FutureTask(new Callable() { + @Override + public String call() throws Exception { + return null; + } + })); + + for (FutureTask processedItem : processedItems) { + taskExecutor.execute(processedItem); + } + + writer.write(processedItems); + + assertEquals(1, writtenItems.size()); + assertTrue(writtenItems.contains("foo")); + } + + private class ListItemWriter implements ItemWriter { + + protected List items; + + public ListItemWriter(List items) { + this.items = items; + } + + @Override + public void write(List items) throws Exception { + this.items.addAll(items); + } + } +}