diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
index 75e3551551..161c6c8b47 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
@@ -16,22 +16,19 @@
 
 package org.springframework.integration.splitter;
 
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.UUID;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.support.MessageBuilder;
 
+import java.util.*;
+
 /**
  * Base class for Message-splitting handlers.
  * 
  * @author Mark Fisher
  * @author Dave Syer
+ * @author Iwein Fuld
  */
 public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
 
@@ -62,6 +59,10 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
 		List<MessageBuilder<?>> messageBuilders = new ArrayList<MessageBuilder<?>>();
 		if (result instanceof Collection) {
 			Collection<?> items = (Collection<?>) result;
+			//TODO put this return statement in a more obvious place
+			if(items.isEmpty()){
+				return null;
+			}
 			int sequenceNumber = 0;
 			int sequenceSize = items.size();
 			for (Object item : items) {
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
index 82fb892938..e2326ec28a 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java
@@ -16,23 +16,25 @@
 
 package org.springframework.integration.splitter;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-
-import java.util.Arrays;
-import java.util.List;
-
 import org.junit.Test;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.support.MessageBuilder;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static junit.framework.Assert.assertEquals;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.*;
+
 /**
  * @author Mark Fisher
+ * @author Iwein Fuld
  */
 public class DefaultSplitterTests {
 
@@ -92,4 +94,15 @@ public class DefaultSplitterTests {
 		assertEquals(message.getHeaders().getId(), reply.getHeaders().getCorrelationId());
 	}
 
+	@Test
+	public void splitMessageWithEmptyCollectionPayload() throws Exception {
+		Message<List<String>> message = MessageBuilder.withPayload(Collections.<String>emptyList()).build();
+		QueueChannel replyChannel = new QueueChannel();
+		DefaultMessageSplitter splitter = new DefaultMessageSplitter();
+		splitter.setOutputChannel(replyChannel);
+		splitter.handleMessage(message);
+		Message<?> output = replyChannel.receive(15);
+		assertThat(output, is(nullValue()));
+	}
+
 }