diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
index fa9b81ac7..7ae445dee 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
@@ -359,7 +359,7 @@ public class FunctionConfiguration {
 				this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, 0);
 				if (this.functionProperties.isComposeFrom()) {
 					SubscribableChannel outputChannel = this.applicationContext.getBean(outputDestinationName, SubscribableChannel.class);
-					//logger.info("Composing at the head of 'output' channel");
+//					logger.info("Composing at the head of 'output' channel");
 					String outputChannelName = ((AbstractMessageChannel) outputChannel).getBeanName();
 					ServiceActivatingHandler handler = createFunctionHandler(function, null, outputChannelName);
 
@@ -510,6 +510,10 @@ public class FunctionConfiguration {
 
 		FunctionWrapper(Function function, ConsumerProperties consumerProperties, ProducerProperties producerProperties) {
 			this.function = function;
+			Type type =  ((FunctionInvocationWrapper) function).getFunctionType();
+			if (FunctionTypeUtils.isReactive(FunctionTypeUtils.getOutputType(type, 0))) {
+				throw new IllegalStateException("Functions with mixed semantics (imperative input vs. reactive output) ar not supported at the moment");
+			}
 			this.consumerProperties = consumerProperties;
 			this.producerProperties = producerProperties;
 			this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
@@ -518,17 +522,15 @@ public class FunctionConfiguration {
 		@SuppressWarnings("unchecked")
 		@Override
 		public Object apply(Message<byte[]> message) {
-
 			if (message != null && consumerProperties != null) {
 				Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils
 						.getField(this.headersField, message.getHeaders());
 				headersMap.put(FunctionProperties.SKIP_CONVERSION_HEADER, consumerProperties.isUseNativeDecoding());
 			}
-
 			Object result = function.apply(message);
-			if (result instanceof Publisher) {
-				throw new IllegalStateException("Routing to functions that return Publisher is not supported "
-						+ "in the context of Spring Cloud Stream.");
+			if (result instanceof Publisher && ((FunctionInvocationWrapper) this.function).getTarget() instanceof RoutingFunction) {
+				throw new IllegalStateException("Routing to functions that return Publisher "
+						+ "is not supported in the context of Spring Cloud Stream.");
 			}
 			return result;
 		}
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
index 3b38050e6..780cd80af 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
@@ -18,6 +18,8 @@ package org.springframework.cloud.stream.function;
 
 import java.io.Serializable;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
@@ -394,6 +396,53 @@ public class ImplicitFunctionBindingTests {
 		}
 	}
 
+	// see https://github.com/spring-cloud/spring-cloud-stream/issues/1896
+	@Test
+	public void testOutputAsCollectionOfMessages() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(SplittableTypesConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcCollectionOfMessages",
+								"--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("aa,bb,cc,dd".getBytes()).build();
+
+			inputDestination.send(inputMessage);
+
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("aa");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("bb");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("cc");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("dd");
+			assertThat(outputDestination.receive(100)).isNull();
+		}
+	}
+
+	@Test
+	public void testOutputAsArrayOfMessages() {
+		System.clearProperty("spring.cloud.function.definition");
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(SplittableTypesConfiguration.class))
+						.web(WebApplicationType.NONE).run("--spring.cloud.function.definition=funcArrayOfMessages",
+								"--spring.jmx.enabled=false")) {
+
+			InputDestination inputDestination = context.getBean(InputDestination.class);
+			OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+			Message<byte[]> inputMessage = MessageBuilder.withPayload("aa,bb,cc,dd".getBytes()).build();
+
+			inputDestination.send(inputMessage);
+
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("aa");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("bb");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("cc");
+			assertThat(new String(outputDestination.receive(100).getPayload())).isEqualTo("dd");
+			assertThat(outputDestination.receive(100)).isNull();
+		}
+	}
+
 	@EnableAutoConfiguration
 	public static class NoEnableBindingConfiguration {
 
@@ -541,15 +590,40 @@ public class ImplicitFunctionBindingTests {
 
 		@Bean
 		public Function<Message<List<Map<String, Integer>>>, Message<List<Map<String, Integer>>>> funcA() {
-			return v -> {
-				return v;
-			};
+			return v -> v;
 		}
 
 		@Bean
 		public Function<Message<List<Map<String, String>>>, Message<List<Map<String, String>>>> funcB() {
+			return v -> v;
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class SplittableTypesConfiguration {
+
+		@Bean
+		public Function<String, Collection<Message<String>>> funcCollectionOfMessages() {
 			return v -> {
-				return v;
+				String[] values = v.split(",");
+				List<Message<String>> messages = new ArrayList<>();
+				for (String value : values) {
+					messages.add(MessageBuilder.withPayload(value).build());
+				}
+				return messages;
+			};
+		}
+
+		@SuppressWarnings("unchecked")
+		@Bean
+		public Function<String, Message<String>[]> funcArrayOfMessages() {
+			return v -> {
+				String[] values = v.split(",");
+				List<Message<String>> messages = new ArrayList<>();
+				for (String value : values) {
+					messages.add(MessageBuilder.withPayload(value).build());
+				}
+				return messages.toArray(new Message[0]);
 			};
 		}
 	}