diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
index f0d4742165..7b4050b666 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
@@ -92,6 +92,10 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 		}
 	}
 
+	protected boolean shouldSplitReplies() {
+		return false;
+	}
+
 	/**
 	 * {@inheritDoc}
 	 */
@@ -111,7 +115,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
 		}
 		MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.channelResolver);
 		MessageHeaders requestHeaders = message.getHeaders();
-		if (result instanceof Iterable) {
+		if (this.shouldSplitReplies() && result instanceof Iterable) {
 			this.sendReplyMessages((Iterable) result, requestHeaders, replyChannel);
 		}
 		else {
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
index 220df43eed..1246e4a55e 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
@@ -33,6 +33,11 @@ import org.springframework.integration.message.MessageBuilder;
  */
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
 
+	@Override
+	protected boolean shouldSplitReplies() {
+		return true;
+	}
+
 	@Override
 	protected final Object handleRequestMessage(Message<?> message) {
 		Object result = this.splitMessage(message);
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/CollectionAndArrayTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CollectionAndArrayTests.java
new file mode 100644
index 0000000000..d5b9ef59c5
--- /dev/null
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CollectionAndArrayTests.java
@@ -0,0 +1,161 @@
+/*
+ * Copyright 2002-2009 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.integration.handler;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.Test;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.integration.splitter.AbstractMessageSplitter;
+
+/**
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class CollectionAndArrayTests {
+
+	@Test
+	public void listWithRequestReplyHandler() {
+		MessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				return Arrays.asList(new String[] { "foo", "bar" });
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNull(reply2);
+		assertTrue(List.class.isAssignableFrom(reply1.getPayload().getClass()));
+		assertEquals(2, ((List<?>) reply1.getPayload()).size());
+	}
+
+	@Test
+	public void setWithRequestReplyHandler() {
+		MessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				return new HashSet<String>(Arrays.asList(new String[] { "foo", "bar" }));
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNull(reply2);
+		assertTrue(Set.class.isAssignableFrom(reply1.getPayload().getClass()));
+		assertEquals(2, ((Set<?>) reply1.getPayload()).size());
+	}
+
+	@Test
+	public void arrayWithRequestReplyHandler() {
+		MessageHandler handler = new AbstractReplyProducingMessageHandler() {
+			@Override
+			protected Object handleRequestMessage(Message<?> requestMessage) {
+				return new String[] { "foo", "bar" };
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNull(reply2);
+		assertTrue(reply1.getPayload().getClass().isArray());
+		assertEquals(2, ((String[]) reply1.getPayload()).length);
+	}
+
+	@Test
+	public void listWithSplittingHandler() {
+		AbstractMessageSplitter handler = new AbstractMessageSplitter() {
+			@Override
+			protected Object splitMessage(Message<?> message) {
+				return Arrays.asList(new String[] { "foo", "bar" });
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNotNull(reply2);
+		assertEquals(String.class, reply1.getPayload().getClass());
+		assertEquals(String.class, reply2.getPayload().getClass());
+		assertEquals("foo", reply1.getPayload());
+		assertEquals("bar", reply2.getPayload());
+	}
+
+	@Test
+	public void setWithSplittingHandler() {
+		AbstractMessageSplitter handler = new AbstractMessageSplitter() {
+			@Override
+			protected Object splitMessage(Message<?> message) {
+				return new HashSet<String>(Arrays.asList(new String[] { "foo", "bar" }));
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNotNull(reply2);
+		assertEquals(String.class, reply1.getPayload().getClass());
+		assertEquals(String.class, reply2.getPayload().getClass());
+	}
+
+	@Test
+	public void arrayWithSplittingHandler() {
+		AbstractMessageSplitter handler = new AbstractMessageSplitter() {
+			@Override
+			protected Object splitMessage(Message<?> message) {
+				return new String[] { "foo", "bar" };
+			}
+		};
+		QueueChannel channel = new QueueChannel();
+		Message<?> message = MessageBuilder.withPayload("test").setReplyChannel(channel).build();
+		handler.handleMessage(message);
+		Message<?> reply1 = channel.receive(0);
+		Message<?> reply2 = channel.receive(0);
+		assertNotNull(reply1);
+		assertNotNull(reply2);
+		assertEquals(String.class, reply1.getPayload().getClass());
+		assertEquals(String.class, reply2.getPayload().getClass());
+		assertEquals("foo", reply1.getPayload());
+		assertEquals("bar", reply2.getPayload());
+	}
+
+}